Separate the NativeRealmMigrator and the NativeUsersStore (elastic/elasticsearch#4932)

This commit removes the NativeRealmMigrator's dependency on the NativeUserStore and instead directly uses the
InternalClient for the migration operations. There are pros and cons to doing it both ways, but I feel this
method makes it more explicit that this is what the migrator is going to do. The downside here is that there
are two places in the code that need to know the inner details of how we store users.

Additionally, by doing this we avoid a race condition between the NativeUsersStore starting and the
NativeRealmMigrator attempting to get all of the reserved users. This race causes the
OldSecurityIndexBackwardsCompatibility tests to fail intermittently.

Original commit: elastic/x-pack-elasticsearch@6c388db535

elasticsearch/src/main/java/org/elasticsearch/xpack/security/SecurityLifecycleService.java

@@ -54,7 +54,7 @@ public class SecurityLifecycleService extends AbstractComponent implements Clust
         clusterService.addListener(this);
         clusterService.addListener(nativeUserStore);
         clusterService.addListener(nativeRolesStore);
-        final NativeRealmMigrator nativeRealmMigrator = new NativeRealmMigrator(settings, nativeUserStore, licenseState);
+        final NativeRealmMigrator nativeRealmMigrator = new NativeRealmMigrator(settings, licenseState, client);
         clusterService.addListener(new SecurityTemplateService(settings, client, nativeRealmMigrator));
         clusterService.addLifecycleListener(new LifecycleListener() {
 

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/esnative/NativeRealmMigrator.java

@@ -7,23 +7,35 @@ package org.elasticsearch.xpack.security.authc.esnative;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
+import org.elasticsearch.action.update.UpdateResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.Requests;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.inject.internal.Nullable;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.xpack.common.GroupedActionListener;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.security.InternalClient;
 import org.elasticsearch.xpack.security.SecurityTemplateService;
-import org.elasticsearch.xpack.security.action.user.ChangePasswordRequest;
 import org.elasticsearch.xpack.security.authc.support.Hasher;
+import org.elasticsearch.xpack.security.authc.support.SecuredString;
+import org.elasticsearch.xpack.security.client.SecurityClient;
 import org.elasticsearch.xpack.security.user.LogstashSystemUser;
+import org.elasticsearch.xpack.security.user.User;
+import org.elasticsearch.xpack.security.user.User.Fields;
 
 /**
  * Performs migration steps for the {@link NativeRealm} and {@link ReservedRealm}.
@@ -34,14 +46,14 @@ import org.elasticsearch.xpack.security.user.LogstashSystemUser;
  */
 public class NativeRealmMigrator {
 
-    private final NativeUsersStore nativeUsersStore;
     private final XPackLicenseState licenseState;
     private final Logger logger;
+    private Client client;
 
-    public NativeRealmMigrator(Settings settings, NativeUsersStore nativeUsersStore, XPackLicenseState licenseState) {
+    public NativeRealmMigrator(Settings settings, XPackLicenseState licenseState, InternalClient internalClient) {
-        this.nativeUsersStore = nativeUsersStore;
         this.licenseState = licenseState;
         this.logger = Loggers.getLogger(getClass(), settings);
+        this.client = internalClient;
     }
 
     /**
@@ -78,7 +90,7 @@ public class NativeRealmMigrator {
     private List<BiConsumer<Version, ActionListener<Void>>> collectUpgradeTasks(@Nullable Version previousVersion) {
         List<BiConsumer<Version, ActionListener<Void>>> tasks = new ArrayList<>();
         if (shouldDisableLogstashUser(previousVersion)) {
-            tasks.add(this::doDisableLogstashUser);
+            tasks.add(this::createLogstashUserAsDisabled);
         }
         if (shouldConvertDefaultPasswords(previousVersion)) {
             tasks.add(this::doConvertDefaultPasswords);
@@ -95,13 +107,35 @@ public class NativeRealmMigrator {
         return previousVersion != null && previousVersion.before(LogstashSystemUser.DEFINED_SINCE);
     }
 
-    private void doDisableLogstashUser(@Nullable Version previousVersion, ActionListener<Void> listener) {
+    private void createLogstashUserAsDisabled(@Nullable Version previousVersion, ActionListener<Void> listener) {
         logger.info("Upgrading security from version [{}] - new reserved user [{}] will default to disabled",
                 previousVersion, LogstashSystemUser.NAME);
         // Only clear the cache is authentication is allowed by the current license
         // otherwise the license management checks will prevent it from completing successfully.
         final boolean clearCache = licenseState.isAuthAllowed();
-        nativeUsersStore.ensureReservedUserIsDisabled(LogstashSystemUser.NAME, clearCache, listener);
+        client.prepareGet(SecurityTemplateService.SECURITY_INDEX_NAME, NativeUsersStore.RESERVED_USER_DOC_TYPE, LogstashSystemUser.NAME)
+                .execute(ActionListener.wrap(getResponse -> {
+                    if (getResponse.isExists()) {
+                        // the document exists - we shouldn't do anything
+                        listener.onResponse(null);
+                    } else {
+                        client.prepareIndex(SecurityTemplateService.SECURITY_INDEX_NAME, NativeUsersStore.RESERVED_USER_DOC_TYPE,
+                                LogstashSystemUser.NAME)
+                                .setSource(Requests.INDEX_CONTENT_TYPE,
+                                        User.Fields.ENABLED.getPreferredName(), false,
+                                        User.Fields.PASSWORD.getPreferredName(), "")
+                                .setCreate(true)
+                                .execute(ActionListener.wrap(r -> {
+                                    if (clearCache) {
+                                        new SecurityClient(client).prepareClearRealmCache()
+                                                .usernames(LogstashSystemUser.NAME)
+                                                .execute(ActionListener.wrap(re -> listener.onResponse(null), listener::onFailure));
+                                    } else {
+                                        listener.onResponse(null);
+                                    }
+                                }, listener::onFailure));
+                    }
+                }, listener::onFailure));
     }
 
     /**
@@ -116,16 +150,25 @@ public class NativeRealmMigrator {
 
     @SuppressWarnings("unused")
     private void doConvertDefaultPasswords(@Nullable Version previousVersion, ActionListener<Void> listener) {
-        nativeUsersStore.getAllReservedUserInfo(ActionListener.wrap(
+        client.prepareSearch(SecurityTemplateService.SECURITY_INDEX_NAME)
-                users -> {
+                .setTypes(NativeUsersStore.RESERVED_USER_DOC_TYPE)
-                    final List<String> toConvert = users.entrySet().stream()
+                .setQuery(QueryBuilders.matchAllQuery())
-                            .filter(entry -> hasOldStyleDefaultPassword(entry.getValue()))
+                .setFetchSource(true)
-                            .map(entry -> entry.getKey())
+                .execute(ActionListener.wrap(searchResponse -> {
-                            .collect(Collectors.toList());
+                    assert searchResponse.getHits().getTotalHits() <= 10 : "there are more than 10 reserved users we need to change " +
+                            "this to retrieve them all!";
+                    Set<String> toConvert = new HashSet<>();
+                    for (SearchHit searchHit : searchResponse.getHits()) {
+                        Map<String, Object> sourceMap = searchHit.getSourceAsMap();
+                        if (hasOldStyleDefaultPassword(sourceMap)) {
+                            toConvert.add(searchHit.getId());
+                        }
+                    }
+
                     if (toConvert.isEmpty()) {
                         listener.onResponse(null);
                     } else {
-                        GroupedActionListener countDownListener = new GroupedActionListener(
+                        GroupedActionListener<UpdateResponse> countDownListener = new GroupedActionListener<>(
                                 ActionListener.wrap((r) -> listener.onResponse(null), listener::onFailure),
                                 toConvert.size(), Collections.emptyList()
                         );
@@ -133,29 +176,30 @@ public class NativeRealmMigrator {
                             logger.debug(
                                     "Upgrading security from version [{}] - marking reserved user [{}] as having default password",
                                     previousVersion, username);
-                            resetReservedUserPassword(username, countDownListener);
+                            client.prepareUpdate(
+                                    SecurityTemplateService.SECURITY_INDEX_NAME,NativeUsersStore.RESERVED_USER_DOC_TYPE, username)
+                                    .setDoc(Fields.PASSWORD.getPreferredName(), "")
+                                    .setRefreshPolicy(RefreshPolicy.IMMEDIATE)
+                                    .execute(countDownListener);
                         });
                     }
-                }, listener::onFailure)
+                }, listener::onFailure));
-        );
     }
 
     /**
-     * Determines whether the supplied {@link NativeUsersStore.ReservedUserInfo} has its password set to be the default password, without
+     * Determines whether the supplied source as a {@link Map} has its password explicitly set to be the default password
-     * having the {@link NativeUsersStore.ReservedUserInfo#hasDefaultPassword} flag set.
      */
-    private boolean hasOldStyleDefaultPassword(NativeUsersStore.ReservedUserInfo userInfo) {
+    private boolean hasOldStyleDefaultPassword(Map<String, Object> userSource) {
-        return userInfo.hasDefaultPassword == false && Hasher.BCRYPT.verify(ReservedRealm.DEFAULT_PASSWORD_TEXT, userInfo.passwordHash);
+        final String passwordHash = (String) userSource.get(User.Fields.PASSWORD.getPreferredName());
-    }
+        if (passwordHash == null) {
+            throw new IllegalStateException("passwordHash should never be null");
+        } else if (passwordHash.isEmpty()) {
+            // we know empty is the new style
+            return false;
+        }
 
-    /**
+        try (SecuredString securedString = new SecuredString(passwordHash.toCharArray())) {
-     * Sets a reserved user's password back to blank, so that the default password functionality applies.
+            return Hasher.BCRYPT.verify(ReservedRealm.DEFAULT_PASSWORD_TEXT, securedString.internalChars());
-     */
+        }
-    void resetReservedUserPassword(String username, ActionListener<Void> listener) {
-        final ChangePasswordRequest request = new ChangePasswordRequest();
-        request.username(username);
-        request.passwordHash(new char[0]);
-        nativeUsersStore.changePassword(request, true, listener);
     }
-
 }

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/esnative/NativeUsersStore.java

@@ -17,8 +17,6 @@ import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.action.search.ClearScrollRequest;
-import org.elasticsearch.action.search.ClearScrollResponse;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
@@ -41,13 +39,11 @@ import org.elasticsearch.index.engine.DocumentMissingException;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
-import org.elasticsearch.xpack.common.GroupedActionListener;
 import org.elasticsearch.xpack.security.InternalClient;
 import org.elasticsearch.xpack.security.SecurityTemplateService;
 import org.elasticsearch.xpack.security.action.realm.ClearRealmCacheRequest;
 import org.elasticsearch.xpack.security.action.realm.ClearRealmCacheResponse;
 import org.elasticsearch.xpack.security.action.user.ChangePasswordRequest;
-import org.elasticsearch.xpack.security.action.user.ChangePasswordRequestBuilder;
 import org.elasticsearch.xpack.security.action.user.DeleteUserRequest;
 import org.elasticsearch.xpack.security.action.user.PutUserRequest;
 import org.elasticsearch.xpack.security.authc.support.Hasher;
@@ -212,15 +208,6 @@ public class NativeUsersStore extends AbstractComponent implements ClusterStateL
      * with a hash of the provided password.
      */
     public void changePassword(final ChangePasswordRequest request, final ActionListener<Void> listener) {
-        changePassword(request, false, listener);
-    }
-
-    /**
-     * This version of {@link #changePassword(ChangePasswordRequest, ActionListener)} exists to that the {@link NativeRealmMigrator}
-     * can force change passwords before the security mapping is {@link #canWrite ready for writing}
-     * @param forceWrite If <code>true</code>, allow the change to take place even if the store is currently read-only.
-     */
-    void changePassword(final ChangePasswordRequest request, boolean forceWrite, final ActionListener<Void> listener) {
         final String username = request.username();
         assert SystemUser.NAME.equals(username) == false && XPackUser.NAME.equals(username) == false : username + "is internal!";
         if (state() != State.STARTED) {
@@ -229,7 +216,7 @@ public class NativeUsersStore extends AbstractComponent implements ClusterStateL
         } else if (isTribeNode) {
             listener.onFailure(new UnsupportedOperationException("users may not be created or modified using a tribe node"));
             return;
-        } else if (canWrite == false && forceWrite == false) {
+        } else if (canWrite == false) {
             listener.onFailure(new IllegalStateException("password cannot be changed as user service cannot write until template and " +
                     "mappings are up to date"));
             return;
@@ -444,16 +431,6 @@ public class NativeUsersStore extends AbstractComponent implements ClusterStateL
         }
     }
 
-    void ensureReservedUserIsDisabled(final String username, boolean clearCache, final ActionListener<Void> listener) {
-        getReservedUserInfo(username, ActionListener.wrap(userInfo -> {
-            if (userInfo == null || userInfo.enabled) {
-                setReservedUserEnabled(username, false, RefreshPolicy.IMMEDIATE, clearCache, listener);
-            } else {
-                listener.onResponse(null);
-            }
-        }, listener::onFailure));
-    }
-
     private void setReservedUserEnabled(final String username, final boolean enabled, final RefreshPolicy refreshPolicy,
                                         boolean clearCache, final ActionListener<Void> listener) {
         try {
@@ -677,12 +654,14 @@ public class NativeUsersStore extends AbstractComponent implements ClusterStateL
                             Map<String, Object> sourceMap = searchHit.getSourceAsMap();
                             String password = (String) sourceMap.get(User.Fields.PASSWORD.getPreferredName());
                             Boolean enabled = (Boolean) sourceMap.get(Fields.ENABLED.getPreferredName());
-                            if (password == null || password.isEmpty()) {
+                            if (password == null) {
-                                listener.onFailure(new IllegalStateException("password hash must not be empty!"));
+                                listener.onFailure(new IllegalStateException("password hash must not be null!"));
-                                break;
+                                return;
                             } else if (enabled == null) {
                                 listener.onFailure(new IllegalStateException("enabled must not be null!"));
-                                break;
+                                return;
+                            } else if (password.isEmpty()) {
+                                userInfos.put(searchHit.getId(), new ReservedUserInfo(ReservedRealm.DEFAULT_PASSWORD_HASH, enabled, true));
                             } else {
                                 userInfos.put(searchHit.getId(), new ReservedUserInfo(password.toCharArray(), enabled, false));
                             }
@@ -703,22 +682,6 @@ public class NativeUsersStore extends AbstractComponent implements ClusterStateL
                 });
     }
 
-    private void clearScrollResponse(String scrollId) {
-        ClearScrollRequest clearScrollRequest = client.prepareClearScroll().addScrollId(scrollId).request();
-        client.clearScroll(clearScrollRequest, new ActionListener<ClearScrollResponse>() {
-            @Override
-            public void onResponse(ClearScrollResponse response) {
-                // cool, it cleared, we don't really care though...
-            }
-
-            @Override
-            public void onFailure(Exception t) {
-                // Not really much to do here except for warn about it...
-                logger.warn((Supplier<?>) () -> new ParameterizedMessage("failed to clear scroll [{}]", scrollId), t);
-            }
-        });
-    }
-
     private <Response> void clearRealmCache(String username, ActionListener<Response> listener, Response response) {
         SecurityClient securityClient = new SecurityClient(client);
         ClearRealmCacheRequest request = securityClient.prepareClearRealmCache()

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authc/esnative/NativeRealmMigratorTests.java

@@ -5,41 +5,62 @@
  */
 package org.elasticsearch.xpack.security.authc.esnative;
 
-import java.util.Arrays;
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ExecutionException;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.get.GetAction;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexAction;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchAction;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.update.UpdateAction;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.index.get.GetResult;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.xpack.security.action.user.ChangePasswordRequest;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xpack.security.InternalClient;
+import org.elasticsearch.xpack.security.SecurityTemplateService;
+import org.elasticsearch.xpack.security.action.realm.ClearRealmCacheAction;
+import org.elasticsearch.xpack.security.action.realm.ClearRealmCacheRequest;
 import org.elasticsearch.xpack.security.authc.support.Hasher;
+import org.elasticsearch.xpack.security.crypto.CryptoService;
 import org.elasticsearch.xpack.security.user.ElasticUser;
 import org.elasticsearch.xpack.security.user.KibanaUser;
 import org.elasticsearch.xpack.security.user.LogstashSystemUser;
+import org.elasticsearch.xpack.security.user.User;
 import org.junit.Before;
 import org.mockito.ArgumentCaptor;
-import org.mockito.Mockito;
 
+import static java.util.Collections.emptyMap;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyBoolean;
-import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.isNull;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -48,44 +69,87 @@ import static org.mockito.Mockito.when;
 
 public class NativeRealmMigratorTests extends ESTestCase {
 
-    private Consumer<ActionListener<Void>> ensureDisabledHandler;
+    private Map<String, Map<String, Object>> reservedUsers;
-    private Map<String, NativeUsersStore.ReservedUserInfo> reservedUsers;
+    private InternalClient internalClient;
-    private NativeUsersStore nativeUsersStore;
+    private Client mockClient;
     private NativeRealmMigrator migrator;
     private XPackLicenseState licenseState;
 
     @Before
-    public void setupMocks() {
+    public void setupMocks() throws IOException {
         final boolean allowClearCache = randomBoolean();
+        mockClient = mock(Client.class);
+        ThreadPool threadPool = mock(ThreadPool.class);
+        when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY));
 
-        ensureDisabledHandler = listener -> listener.onResponse(null);
+        internalClient = new InternalClient(Settings.EMPTY, threadPool, mockClient,
-        nativeUsersStore = Mockito.mock(NativeUsersStore.class);
+                new CryptoService(Settings.EMPTY, new Environment(Settings.builder().put("path.home", createTempDir()).build())));
-        Mockito.doAnswer(invocation -> {
+        doAnswer(invocationOnMock -> {
-            ActionListener<Void> listener = (ActionListener<Void>) invocation.getArguments()[2];
+            SearchRequest request = (SearchRequest) invocationOnMock.getArguments()[1];
-            ensureDisabledHandler.accept(listener);
+            ActionListener listener = (ActionListener) invocationOnMock.getArguments()[2];
-            return null;
+            if (request.indices().length == 1 && request.indices()[0].equals(SecurityTemplateService.SECURITY_INDEX_NAME)) {
-        }).when(nativeUsersStore).ensureReservedUserIsDisabled(anyString(), eq(allowClearCache), any(ActionListener.class));
+                SearchResponse response = new SearchResponse() {
+                    @Override
+                    public SearchHits getHits() {
+                        List<SearchHit> hits = reservedUsers.entrySet().stream()
+                                .map((info) -> {
+                                    SearchHit hit = new SearchHit(randomInt(), info.getKey(), null, null, emptyMap());
+                                    try {
+                                        hit.sourceRef(JsonXContent.contentBuilder().map(info.getValue()).bytes());
+                                    } catch (IOException e) {
+                                        throw new UncheckedIOException(e);
+                                    }
+                                    return hit;
+                                }).collect(Collectors.toList());
+                        return new SearchHits(hits.toArray(new SearchHit[0]), (long) reservedUsers.size(), 0.0f);
+                    }
+                };
+                listener.onResponse(response);
+            } else {
+                listener.onResponse(null);
+            }
+            return Void.TYPE;
+        }).when(mockClient).execute(eq(SearchAction.INSTANCE), any(SearchRequest.class), any(ActionListener.class));
 
-        Mockito.doAnswer(invocation -> {
+        doAnswer(invocationOnMock -> {
-            ActionListener<Void> listener = (ActionListener<Void>) invocation.getArguments()[2];
+            GetRequest request = (GetRequest) invocationOnMock.getArguments()[1];
+            ActionListener listener = (ActionListener) invocationOnMock.getArguments()[2];
+            if (request.indices().length == 1 && request.indices()[0].equals(SecurityTemplateService.SECURITY_INDEX_NAME)
+                    && request.type().equals(NativeUsersStore.RESERVED_USER_DOC_TYPE)) {
+                final boolean exists = reservedUsers.get(request.id()) != null;
+                GetResult getResult = new GetResult(SecurityTemplateService.SECURITY_INDEX_NAME, NativeUsersStore.RESERVED_USER_DOC_TYPE,
+                        request.id(), randomLong(), exists, JsonXContent.contentBuilder().map(reservedUsers.get(request.id())).bytes(),
+                        emptyMap());
+                listener.onResponse(new GetResponse(getResult));
+            } else {
+                listener.onResponse(null);
+            }
+            return Void.TYPE;
+        }).when(mockClient).execute(eq(GetAction.INSTANCE), any(GetRequest.class), any(ActionListener.class));
+
+        doAnswer(invocationOnMock -> {
+            ActionListener listener = (ActionListener) invocationOnMock.getArguments()[2];
             listener.onResponse(null);
-            return null;
+            return Void.TYPE;
-        }).when(nativeUsersStore).changePassword(any(ChangePasswordRequest.class), anyBoolean(), any(ActionListener.class));
+        }).when(mockClient).execute(eq(ClearRealmCacheAction.INSTANCE), any(ClearRealmCacheRequest.class), any(ActionListener.class));
 
-        reservedUsers = Collections.emptyMap();
+        doAnswer(invocationOnMock -> {
-        Mockito.doAnswer(invocation -> {
+            ActionListener listener = (ActionListener) invocationOnMock.getArguments()[2];
-            ActionListener<Map<String, NativeUsersStore.ReservedUserInfo>> listener =
+            listener.onResponse(null);
-                    (ActionListener<Map<String, NativeUsersStore.ReservedUserInfo>>) invocation.getArguments()[0];
+            return Void.TYPE;
-            listener.onResponse(reservedUsers);
+        }).when(mockClient).execute(eq(IndexAction.INSTANCE), any(IndexRequest.class), any(ActionListener.class));
-            return null;
+        doAnswer(invocationOnMock -> {
-        }).when(nativeUsersStore).getAllReservedUserInfo(any(ActionListener.class));
+            ActionListener listener = (ActionListener) invocationOnMock.getArguments()[2];
+            listener.onResponse(null);
+            return Void.TYPE;
+        }).when(mockClient).execute(eq(UpdateAction.INSTANCE), any(UpdateRequest.class), any(ActionListener.class));
 
         final Settings settings = Settings.EMPTY;
 
         licenseState = mock(XPackLicenseState.class);
         when(licenseState.isAuthAllowed()).thenReturn(allowClearCache);
 
-        migrator = new NativeRealmMigrator(settings, nativeUsersStore, licenseState);
+        migrator = new NativeRealmMigrator(settings, licenseState, internalClient);
     }
 
     public void testNoChangeOnFreshInstall() throws Exception {
@@ -99,7 +163,10 @@ public class NativeRealmMigratorTests extends ESTestCase {
     public void testDisableLogstashAndConvertPasswordsOnUpgradeFromVersionPriorToV5_2() throws Exception {
         this.reservedUsers = Collections.singletonMap(
                 KibanaUser.NAME,
-                new NativeUsersStore.ReservedUserInfo(Hasher.BCRYPT.hash(ReservedRealm.DEFAULT_PASSWORD_TEXT), false, false)
+                MapBuilder.<String, Object>newMapBuilder()
+                        .put(User.Fields.PASSWORD.getPreferredName(), new String(Hasher.BCRYPT.hash(ReservedRealm.DEFAULT_PASSWORD_TEXT)))
+                        .put(User.Fields.ENABLED.getPreferredName(), false)
+                        .immutableMap()
         );
         verifyUpgrade(randomFrom(Version.V_5_1_1_UNRELEASED, Version.V_5_0_2, Version.V_5_0_0), true, true);
     }
@@ -107,44 +174,49 @@ public class NativeRealmMigratorTests extends ESTestCase {
     public void testConvertPasswordsOnUpgradeFromVersion5_2() throws Exception {
         this.reservedUsers = randomSubsetOf(randomIntBetween(0, 3), LogstashSystemUser.NAME, KibanaUser.NAME, ElasticUser.NAME)
                 .stream().collect(Collectors.toMap(Function.identity(),
-                        name -> new NativeUsersStore.ReservedUserInfo(Hasher.BCRYPT.hash(ReservedRealm.DEFAULT_PASSWORD_TEXT),
+                        name -> MapBuilder.<String, Object>newMapBuilder()
-                                randomBoolean(), false)
+                                .put(User.Fields.PASSWORD.getPreferredName(),
+                                        new String(Hasher.BCRYPT.hash(ReservedRealm.DEFAULT_PASSWORD_TEXT)))
+                                .put(User.Fields.ENABLED.getPreferredName(), randomBoolean())
+                                .immutableMap()
                 ));
         verifyUpgrade(Version.V_5_2_0_UNRELEASED, false, true);
     }
 
-    public void testExceptionInUsersStoreIsPropagatedToListener() throws Exception {
-        final RuntimeException thrown = new RuntimeException("Forced failure");
-        this.ensureDisabledHandler = listener -> listener.onFailure(thrown);
-        final PlainActionFuture<Boolean> future = doUpgrade(Version.V_5_0_0);
-        final ExecutionException caught = expectThrows(ExecutionException.class, future::get);
-        assertThat(caught.getCause(), is(thrown));
-    }
-
     private void verifyUpgrade(Version fromVersion, boolean disableLogstashUser, boolean convertDefaultPasswords) throws Exception {
         final PlainActionFuture<Boolean> future = doUpgrade(fromVersion);
         boolean expectedResult = false;
         if (disableLogstashUser) {
             final boolean clearCache = licenseState.isAuthAllowed();
-            verify(nativeUsersStore).ensureReservedUserIsDisabled(eq(LogstashSystemUser.NAME), eq(clearCache), any());
+            ArgumentCaptor<GetRequest> captor = ArgumentCaptor.forClass(GetRequest.class);
+            verify(mockClient).execute(eq(GetAction.INSTANCE), captor.capture(), any(ActionListener.class));
+            assertEquals(LogstashSystemUser.NAME, captor.getValue().id());
+            ArgumentCaptor<IndexRequest> indexCaptor = ArgumentCaptor.forClass(IndexRequest.class);
+            verify(mockClient).execute(eq(IndexAction.INSTANCE), indexCaptor.capture(), any(ActionListener.class));
+            assertEquals(LogstashSystemUser.NAME, indexCaptor.getValue().id());
+            assertEquals(false, indexCaptor.getValue().sourceAsMap().get(User.Fields.ENABLED.getPreferredName()));
+
+            if (clearCache) {
+                verify(mockClient).execute(eq(ClearRealmCacheAction.INSTANCE), any(ClearRealmCacheRequest.class),
+                        any(ActionListener.class));
+            }
             expectedResult = true;
         }
         if (convertDefaultPasswords) {
-            verify(nativeUsersStore).getAllReservedUserInfo(any());
+            verify(mockClient).execute(eq(SearchAction.INSTANCE), any(SearchRequest.class), any(ActionListener.class));
-            ArgumentCaptor<ChangePasswordRequest> captor = ArgumentCaptor.forClass(ChangePasswordRequest.class);
+            ArgumentCaptor<UpdateRequest> captor = ArgumentCaptor.forClass(UpdateRequest.class);
-            verify(nativeUsersStore, times(this.reservedUsers.size()))
+            verify(mockClient, times(this.reservedUsers.size()))
-                    .changePassword(captor.capture(), eq(true), any(ActionListener.class));
+                    .execute(eq(UpdateAction.INSTANCE), captor.capture(), any(ActionListener.class));
-            final List<ChangePasswordRequest> requests = captor.getAllValues();
+            final List<UpdateRequest> requests = captor.getAllValues();
             this.reservedUsers.keySet().forEach(u -> {
-                ChangePasswordRequest request = requests.stream().filter(r -> r.username().equals(u)).findFirst().get();
+                UpdateRequest request = requests.stream().filter(r -> r.id().equals(u)).findFirst().get();
                 assertThat(request.validate(), nullValue(ActionRequestValidationException.class));
-                assertThat(request.username(), equalTo(u));
+                assertThat(request.doc().sourceAsMap(), hasEntry(is(User.Fields.PASSWORD.getPreferredName()), is("")));
-                assertThat(request.passwordHash().length, equalTo(0));
                 assertThat(request.getRefreshPolicy(), equalTo(WriteRequest.RefreshPolicy.IMMEDIATE));
             });
             expectedResult = true;
         }
-        verifyNoMoreInteractions(nativeUsersStore);
+        verifyNoMoreInteractions(mockClient);
         assertThat(future.get(), is(expectedResult));
     }