Merge branch 'master' into enhancement/use_shard_bulk_for_single_ops

Original commit: elastic/x-pack-elasticsearch@98f4e74d2e

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

@@ -241,7 +241,7 @@ public class Security implements ActionPlugin, IngestPlugin, NetworkPlugin {
         }
 
         List<Object> components = new ArrayList<>();
-        final SecurityContext securityContext = new SecurityContext(settings, threadPool, cryptoService);
+        final SecurityContext securityContext = new SecurityContext(settings, threadPool.getThreadContext(), cryptoService);
         components.add(securityContext);
 
         // realms construction

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

@@ -37,9 +37,9 @@ public class SecurityContext {
      * If cryptoService is null, security is disabled and {@link #getUser()}
      * and {@link #getAuthentication()} will always return null.
      */
-    public SecurityContext(Settings settings, ThreadPool threadPool, CryptoService cryptoService) {
+    public SecurityContext(Settings settings, ThreadContext threadContext, CryptoService cryptoService) {
         this.logger = Loggers.getLogger(getClass(), settings);
-        this.threadContext = threadPool.getThreadContext();
+        this.threadContext = threadContext;
         this.cryptoService = cryptoService;
         this.signUserHeader = AuthenticationService.SIGN_USER_HEADER.get(settings);
         this.nodeName = Node.NODE_NAME_SETTING.get(settings);

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authc/Authentication.java

@@ -25,11 +25,13 @@ public class Authentication {
     private final User user;
     private final RealmRef authenticatedBy;
     private final RealmRef lookedUpBy;
+    private final Version version;
 
     public Authentication(User user, RealmRef authenticatedBy, RealmRef lookedUpBy) {
         this.user = Objects.requireNonNull(user);
         this.authenticatedBy = Objects.requireNonNull(authenticatedBy);
         this.lookedUpBy = lookedUpBy;
+        this.version = Version.CURRENT;
     }
 
     public Authentication(StreamInput in) throws IOException {
@@ -40,6 +42,7 @@ public class Authentication {
         } else {
             this.lookedUpBy = null;
         }
+        this.version = in.getVersion();
     }
 
     public User getUser() {
@@ -70,6 +73,10 @@ public class Authentication {
         return lookedUpBy;
     }
 
+    public Version getVersion() {
+        return version;
+    }
+
     public static Authentication readFromContext(ThreadContext ctx, CryptoService cryptoService, boolean sign)
             throws IOException, IllegalArgumentException {
         Authentication authentication = ctx.getTransient(AUTHENTICATION_KEY);

elasticsearch/src/main/java/org/elasticsearch/xpack/security/authz/store/ReservedRolesStore.java

@@ -16,6 +16,7 @@ import org.elasticsearch.xpack.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.security.authz.permission.Role;
 import org.elasticsearch.xpack.security.support.MetadataUtils;
 
+import org.elasticsearch.xpack.security.user.KibanaUser;
 import org.elasticsearch.xpack.security.user.SystemUser;
 
 public class ReservedRolesStore {
@@ -53,7 +54,7 @@ public class ReservedRolesStore {
                 .put("reporting_user", new RoleDescriptor("reporting_user", null, new RoleDescriptor.IndicesPrivileges[] {
                         RoleDescriptor.IndicesPrivileges.builder().indices(".reporting-*").privileges("read", "write").build() },
                         null, MetadataUtils.DEFAULT_RESERVED_METADATA))
-                .put("kibana", new RoleDescriptor("kibana", new String[] { "monitor", MonitoringBulkAction.NAME},
+                .put(KibanaUser.ROLE_NAME, new RoleDescriptor(KibanaUser.ROLE_NAME, new String[] { "monitor", MonitoringBulkAction.NAME},
                         new RoleDescriptor.IndicesPrivileges[] {
                             RoleDescriptor.IndicesPrivileges.builder().indices(".kibana*", ".reporting-*").privileges("all").build() },
                         null, MetadataUtils.DEFAULT_RESERVED_METADATA))

elasticsearch/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.security.transport;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.common.CheckedConsumer;
@@ -24,13 +25,16 @@ import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportResponseHandler;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.transport.TransportSettings;
+import org.elasticsearch.xpack.XPackSettings;
 import org.elasticsearch.xpack.security.SecurityContext;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationUtils;
 import org.elasticsearch.xpack.security.authz.accesscontrol.RequestContext;
 import org.elasticsearch.xpack.security.transport.netty4.SecurityNetty4Transport;
+import org.elasticsearch.xpack.security.user.KibanaUser;
 import org.elasticsearch.xpack.security.user.SystemUser;
+import org.elasticsearch.xpack.security.user.User;
 import org.elasticsearch.xpack.ssl.SSLService;
 
 import java.io.IOException;
@@ -56,6 +60,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
     private final ThreadPool threadPool;
     private final Settings settings;
     private final SecurityContext securityContext;
+    private final boolean reservedRealmEnabled;
 
     public SecurityServerTransportInterceptor(Settings settings,
                                               ThreadPool threadPool,
@@ -73,6 +78,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
         this.sslService = sslService;
         this.securityContext = securityContext;
         this.profileFilters = initializeProfileFilters(destructiveOperations);
+        this.reservedRealmEnabled = XPackSettings.RESERVED_REALM_ENABLED_SETTING.get(settings);
     }
 
     @Override
@@ -87,6 +93,13 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
                     if (AuthorizationUtils.shouldReplaceUserWithSystem(threadPool.getThreadContext(), action)) {
                         securityContext.executeAsUser(SystemUser.INSTANCE, (original) -> sendWithUser(connection, action, request, options,
                                 new ContextRestoreResponseHandler<>(threadPool.getThreadContext(), original, handler), sender));
+                    } else if (reservedRealmEnabled && connection.getVersion().before(Version.V_5_2_0_UNRELEASED) &&
+                            KibanaUser.NAME.equals(securityContext.getUser().principal())) {
+                        final User kibanaUser = securityContext.getUser();
+                        final User bwcKibanaUser = new User(kibanaUser.principal(), new String[] { "kibana" }, kibanaUser.fullName(),
+                                kibanaUser.email(), kibanaUser.metadata(), kibanaUser.enabled());
+                        securityContext.executeAsUser(bwcKibanaUser, (original) -> sendWithUser(connection, action, request, options,
+                                new ContextRestoreResponseHandler<>(threadPool.getThreadContext(), original, handler), sender));
                     } else {
                         sendWithUser(connection, action, request, options, handler, sender);
                     }
@@ -134,11 +147,13 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
             switch (type) {
                 case "client":
                     profileFilters.put(entry.getKey(), new ServerTransportFilter.ClientProfile(authcService, authzService,
-                            threadPool.getThreadContext(), extractClientCert, destructiveOperations));
+                            threadPool.getThreadContext(), extractClientCert, destructiveOperations, reservedRealmEnabled,
+                            securityContext));
                     break;
                 default:
                     profileFilters.put(entry.getKey(), new ServerTransportFilter.NodeProfile(authcService, authzService,
-                            threadPool.getThreadContext(), extractClientCert, destructiveOperations));
+                            threadPool.getThreadContext(), extractClientCert, destructiveOperations, reservedRealmEnabled,
+                            securityContext));
             }
         }
 
@@ -147,7 +162,7 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
             final boolean clientAuth = sslService.isSSLClientAuthEnabled(transportSSLSettings);
             final boolean extractClientCert = profileSsl && clientAuth;
             profileFilters.put(TransportSettings.DEFAULT_PROFILE, new ServerTransportFilter.NodeProfile(authcService, authzService,
-                    threadPool.getThreadContext(), extractClientCert, destructiveOperations));
+                    threadPool.getThreadContext(), extractClientCert, destructiveOperations, reservedRealmEnabled, securityContext));
         }
 
         return Collections.unmodifiableMap(profileFilters);

elasticsearch/src/main/java/org/elasticsearch/xpack/security/transport/ServerTransportFilter.java

@@ -10,6 +10,7 @@ import io.netty.handler.ssl.SslHandler;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
@@ -22,11 +23,16 @@ import org.elasticsearch.transport.DelegatingTransportChannel;
 import org.elasticsearch.transport.TcpTransportChannel;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.security.SecurityContext;
 import org.elasticsearch.xpack.security.action.SecurityActionMapper;
+import org.elasticsearch.xpack.security.authc.Authentication;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authc.pki.PkiRealm;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationUtils;
+import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.ContextRestoreResponseHandler;
+import org.elasticsearch.xpack.security.user.KibanaUser;
+import org.elasticsearch.xpack.security.user.User;
 
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLPeerUnverifiedException;
@@ -65,14 +71,19 @@ public interface ServerTransportFilter {
         private final ThreadContext threadContext;
         private final boolean extractClientCert;
         private final DestructiveOperations destructiveOperations;
+        private final boolean reservedRealmEnabled;
+        private final SecurityContext securityContext;
 
-        public NodeProfile(AuthenticationService authcService, AuthorizationService authzService,
+        NodeProfile(AuthenticationService authcService, AuthorizationService authzService,
-                           ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations) {
+                    ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations,
+                    boolean reservedRealmEnabled, SecurityContext securityContext) {
             this.authcService = authcService;
             this.authzService = authzService;
             this.threadContext = threadContext;
             this.extractClientCert = extractClientCert;
             this.destructiveOperations = destructiveOperations;
+            this.reservedRealmEnabled = reservedRealmEnabled;
+            this.securityContext = securityContext;
         }
 
         @Override
@@ -112,12 +123,31 @@ public interface ServerTransportFilter {
             }
 
             authcService.authenticate(securityAction, request, null, ActionListener.wrap((authentication) -> {
+                    if (reservedRealmEnabled && authentication.getVersion().before(Version.V_5_2_0_UNRELEASED)
+                            && KibanaUser.NAME.equals(authentication.getUser().principal())) {
+                        // the authentication came from an older node - so let's replace the user with our version
+                        final User kibanaUser = new KibanaUser(authentication.getUser().enabled());
+                        if (kibanaUser.enabled()) {
+                            securityContext.executeAsUser(kibanaUser, (original) -> {
+                                final Authentication replacedUserAuth = Authentication.getAuthentication(threadContext);
+                                final AuthorizationUtils.AsyncAuthorizer asyncAuthorizer =
+                                        new AuthorizationUtils.AsyncAuthorizer(replacedUserAuth, listener, (userRoles, runAsRoles) -> {
+                                            authzService.authorize(replacedUserAuth, securityAction, request, userRoles, runAsRoles);
+                                            listener.onResponse(null);
+                                        });
+                                asyncAuthorizer.authorize(authzService);
+                            });
+                        } else {
+                            throw new IllegalStateException("a disabled user should never be sent. " + kibanaUser);
+                        }
+                    } else {
                         final AuthorizationUtils.AsyncAuthorizer asyncAuthorizer =
                                 new AuthorizationUtils.AsyncAuthorizer(authentication, listener, (userRoles, runAsRoles) -> {
                                     authzService.authorize(authentication, securityAction, request, userRoles, runAsRoles);
                                     listener.onResponse(null);
                                 });
                         asyncAuthorizer.authorize(authzService);
+                    }
                 }, listener::onFailure));
         }
     }
@@ -151,9 +181,11 @@ public interface ServerTransportFilter {
      */
     class ClientProfile extends NodeProfile {
 
-        public ClientProfile(AuthenticationService authcService, AuthorizationService authzService,
+        ClientProfile(AuthenticationService authcService, AuthorizationService authzService,
-                             ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations) {
+                             ThreadContext threadContext, boolean extractClientCert, DestructiveOperations destructiveOperations,
-            super(authcService, authzService, threadContext, extractClientCert, destructiveOperations);
+                             boolean reservedRealmEnabled, SecurityContext securityContext) {
+            super(authcService, authzService, threadContext, extractClientCert, destructiveOperations, reservedRealmEnabled,
+                    securityContext);
         }
 
         @Override

elasticsearch/src/main/java/org/elasticsearch/xpack/security/user/KibanaUser.java

@@ -13,7 +13,7 @@ import org.elasticsearch.xpack.security.support.MetadataUtils;
 public class KibanaUser extends User {
 
     public static final String NAME = "kibana";
-    public static final String ROLE_NAME = "kibana";
+    public static final String ROLE_NAME = "kibana_system";
 
     public KibanaUser(boolean enabled) {
         super(NAME, new String[]{ ROLE_NAME }, null, null, MetadataUtils.DEFAULT_RESERVED_METADATA, enabled);

elasticsearch/src/main/java/org/elasticsearch/xpack/watcher/support/search/WatcherSearchTemplateService.java

@@ -75,7 +75,7 @@ public class WatcherSearchTemplateService extends AbstractComponent {
         if (source != null && source.length() > 0) {
             try (XContentParser parser = XContentFactory.xContent(source).createParser(xContentRegistry, source)) {
                 sourceBuilder.parseXContent(new QueryParseContext(parser, parseFieldMatcher),
-                        searchRequestParsers.aggParsers, searchRequestParsers.suggesters);
+                        searchRequestParsers.suggesters);
                 searchRequest.source(sourceBuilder);
             }
         }

elasticsearch/src/test/java/org/elasticsearch/xpack/security/SecurityContextTests.java

@@ -10,7 +10,6 @@ import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.security.authc.Authentication;
 import org.elasticsearch.xpack.security.authc.Authentication.RealmRef;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
@@ -22,9 +21,6 @@ import org.junit.Before;
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicReference;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
 public class SecurityContextTests extends ESTestCase {
 
     private boolean signHeader;
@@ -42,9 +38,7 @@ public class SecurityContextTests extends ESTestCase {
                 .build();
         threadContext = new ThreadContext(settings);
         cryptoService = new CryptoService(settings, new Environment(settings));
-        ThreadPool threadPool = mock(ThreadPool.class);
+        securityContext = new SecurityContext(settings, threadContext, cryptoService);
-        when(threadPool.getThreadContext()).thenReturn(threadContext);
-        securityContext = new SecurityContext(settings, threadPool, cryptoService);
     }
 
     public void testGetAuthenticationAndUserInEmptyContext() throws IOException {

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authz/store/FileRolesStoreTests.java

@@ -421,7 +421,7 @@ public class FileRolesStoreTests extends ESTestCase {
         // the system role will always be checked first
         assertThat(events.get(0), containsString("Role [_system] is reserved"));
         assertThat(events.get(1), containsString("Role [superuser] is reserved"));
-        assertThat(events.get(2), containsString("Role [kibana] is reserved"));
+        assertThat(events.get(2), containsString("Role [kibana_system] is reserved"));
         assertThat(events.get(3), containsString("Role [transport_client] is reserved"));
     }
 

elasticsearch/src/test/java/org/elasticsearch/xpack/security/authz/store/ReservedRolesStoreTests.java

@@ -55,7 +55,7 @@ import static org.hamcrest.Matchers.is;
 public class ReservedRolesStoreTests extends ESTestCase {
 
     public void testIsReserved() {
-        assertThat(ReservedRolesStore.isReserved("kibana"), is(true));
+        assertThat(ReservedRolesStore.isReserved("kibana_system"), is(true));
         assertThat(ReservedRolesStore.isReserved("superuser"), is(true));
         assertThat(ReservedRolesStore.isReserved("foobar"), is(false));
         assertThat(ReservedRolesStore.isReserved(SystemUser.ROLE_NAME), is(true));
@@ -92,7 +92,7 @@ public class ReservedRolesStoreTests extends ESTestCase {
     }
 
     public void testKibanaRole() {
-        RoleDescriptor roleDescriptor = new ReservedRolesStore().roleDescriptor("kibana");
+        RoleDescriptor roleDescriptor = new ReservedRolesStore().roleDescriptor("kibana_system");
         assertNotNull(roleDescriptor);
         assertThat(roleDescriptor.getMetadata(), hasEntry("_reserved", true));
 

elasticsearch/src/test/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptorTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.security.transport;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -21,6 +22,7 @@ import org.elasticsearch.transport.TransportRequestOptions;
 import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportResponse.Empty;
 import org.elasticsearch.transport.TransportResponseHandler;
+import org.elasticsearch.xpack.XPackSettings;
 import org.elasticsearch.xpack.security.SecurityContext;
 import org.elasticsearch.xpack.security.authc.Authentication;
 import org.elasticsearch.xpack.security.authc.Authentication.RealmRef;
@@ -28,6 +30,7 @@ import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.crypto.CryptoService;
 import org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor.ContextRestoreResponseHandler;
+import org.elasticsearch.xpack.security.user.KibanaUser;
 import org.elasticsearch.xpack.security.user.SystemUser;
 import org.elasticsearch.xpack.security.user.User;
 import org.elasticsearch.xpack.ssl.SSLService;
@@ -37,10 +40,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
+import static org.hamcrest.Matchers.arrayContaining;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
@@ -63,7 +68,7 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         threadContext = new ThreadContext(settings);
         when(threadPool.getThreadContext()).thenReturn(threadContext);
         cryptoService = new CryptoService(settings, new Environment(settings));
-        securityContext = spy(new SecurityContext(settings, threadPool, cryptoService));
+        securityContext = spy(new SecurityContext(settings, threadPool.getThreadContext(), cryptoService));
         xPackLicenseState = mock(XPackLicenseState.class);
         when(xPackLicenseState.isAuthAllowed()).thenReturn(true);
     }
@@ -112,7 +117,9 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
                 sendingUser.set(securityContext.getUser());
             }
         });
-        sender.sendRequest(null, "indices:foo", null, null, null);
+        Transport.Connection connection = mock(Transport.Connection.class);
+        when(connection.getVersion()).thenReturn(Version.CURRENT);
+        sender.sendRequest(connection, "indices:foo", null, null, null);
         assertTrue(calledWrappedSender.get());
         assertEquals(user, sendingUser.get());
         assertEquals(user, securityContext.getUser());
@@ -168,8 +175,10 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
                 fail("sender should not be called!");
             }
         });
+        Transport.Connection connection = mock(Transport.Connection.class);
+        when(connection.getVersion()).thenReturn(Version.CURRENT);
         IllegalStateException e =
-                expectThrows(IllegalStateException.class, () -> sender.sendRequest(null, "indices:foo", null, null, null));
+                expectThrows(IllegalStateException.class, () -> sender.sendRequest(connection, "indices:foo", null, null, null));
         assertEquals("there should always be a user when sending a message", e.getMessage());
         assertNull(securityContext.getUser());
         verify(xPackLicenseState).isAuthAllowed();
@@ -177,6 +186,66 @@ public class SecurityServerTransportInterceptorTests extends ESTestCase {
         verifyNoMoreInteractions(xPackLicenseState);
     }
 
+    public void testSendWithKibanaUser() throws Exception {
+        final User user = new KibanaUser(true);
+        final Authentication authentication = new Authentication(user, new RealmRef("reserved", "reserved", "node1"), null);
+        authentication.writeToContext(threadContext, cryptoService, AuthenticationService.SIGN_USER_HEADER.get(settings));
+        threadContext.putTransient(AuthorizationService.ORIGINATING_ACTION_KEY, "indices:foo");
+
+        SecurityServerTransportInterceptor interceptor = new SecurityServerTransportInterceptor(settings, threadPool,
+                mock(AuthenticationService.class), mock(AuthorizationService.class), xPackLicenseState, mock(SSLService.class),
+                securityContext, new DestructiveOperations(Settings.EMPTY, new ClusterSettings(Settings.EMPTY,
+                Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))));
+
+        AtomicBoolean calledWrappedSender = new AtomicBoolean(false);
+        AtomicReference<User> sendingUser = new AtomicReference<>();
+        AsyncSender intercepted = new AsyncSender() {
+            @Override
+            public <T extends TransportResponse> void sendRequest(Transport.Connection connection, String action, TransportRequest request,
+                                                                  TransportRequestOptions options, TransportResponseHandler<T> handler) {
+                if (calledWrappedSender.compareAndSet(false, true) == false) {
+                    fail("sender called more than once!");
+                }
+                sendingUser.set(securityContext.getUser());
+            }
+        };
+        AsyncSender sender = interceptor.interceptSender(intercepted);
+        Transport.Connection connection = mock(Transport.Connection.class);
+        when(connection.getVersion()).thenReturn(Version.fromId(randomIntBetween(Version.V_5_0_0_ID, Version.V_5_2_0_ID_UNRELEASED - 100)));
+        sender.sendRequest(connection, "indices:foo[s]", null, null, null);
+        assertTrue(calledWrappedSender.get());
+        assertNotEquals(user, sendingUser.get());
+        assertEquals(KibanaUser.NAME, sendingUser.get().principal());
+        assertThat(sendingUser.get().roles(), arrayContaining("kibana"));
+        assertEquals(user, securityContext.getUser());
+
+        // reset and test with version that was changed
+        calledWrappedSender.set(false);
+        sendingUser.set(null);
+        when(connection.getVersion()).thenReturn(Version.V_5_2_0_UNRELEASED);
+        sender.sendRequest(connection, "indices:foo[s]", null, null, null);
+        assertTrue(calledWrappedSender.get());
+        assertEquals(user, sendingUser.get());
+
+        // reset and disable reserved realm
+        calledWrappedSender.set(false);
+        sendingUser.set(null);
+        when(connection.getVersion()).thenReturn(Version.V_5_0_0);
+        settings = Settings.builder().put(settings).put(XPackSettings.RESERVED_REALM_ENABLED_SETTING.getKey(), false).build();
+        interceptor = new SecurityServerTransportInterceptor(settings, threadPool,
+                mock(AuthenticationService.class), mock(AuthorizationService.class), xPackLicenseState, mock(SSLService.class),
+                securityContext, new DestructiveOperations(Settings.EMPTY, new ClusterSettings(Settings.EMPTY,
+                Collections.singleton(DestructiveOperations.REQUIRES_NAME_SETTING))));
+        sender = interceptor.interceptSender(intercepted);
+        sender.sendRequest(connection, "indices:foo[s]", null, null, null);
+        assertTrue(calledWrappedSender.get());
+        assertEquals(user, sendingUser.get());
+
+        verify(xPackLicenseState, times(3)).isAuthAllowed();
+        verify(securityContext, times(1)).executeAsUser(any(User.class), any(Consumer.class));
+        verifyNoMoreInteractions(xPackLicenseState);
+    }
+
     public void testContextRestoreResponseHandler() throws Exception {
         ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
 

elasticsearch/src/test/java/org/elasticsearch/xpack/security/transport/ServerTransportFilterTests.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.security.transport;
 
 import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.MockIndicesRequest;
 import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
@@ -17,26 +18,33 @@ import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportSettings;
+import org.elasticsearch.xpack.security.SecurityContext;
 import org.elasticsearch.xpack.security.authc.Authentication;
 import org.elasticsearch.xpack.security.authc.Authentication.RealmRef;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.permission.Role;
 import org.elasticsearch.xpack.security.authz.store.ReservedRolesStore;
+import org.elasticsearch.xpack.security.crypto.CryptoService;
+import org.elasticsearch.xpack.security.user.KibanaUser;
 import org.elasticsearch.xpack.security.user.SystemUser;
 import org.elasticsearch.xpack.security.user.User;
 import org.elasticsearch.xpack.security.user.XPackUser;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.mock.orig.Mockito.times;
 import static org.elasticsearch.xpack.security.support.Exceptions.authenticationError;
 import static org.elasticsearch.xpack.security.support.Exceptions.authorizationError;
+import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
@@ -50,6 +58,7 @@ import static org.mockito.Mockito.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
 
 public class ServerTransportFilterTests extends ESTestCase {
+
     private AuthenticationService authcService;
     private AuthorizationService authzService;
     private TransportChannel channel;
@@ -72,6 +81,7 @@ public class ServerTransportFilterTests extends ESTestCase {
     public void testInbound() throws Exception {
         TransportRequest request = mock(TransportRequest.class);
         Authentication authentication = mock(Authentication.class);
+        when(authentication.getVersion()).thenReturn(Version.CURRENT);
         when(authentication.getUser()).thenReturn(SystemUser.INSTANCE);
         when(authentication.getRunAsUser()).thenReturn(SystemUser.INSTANCE);
         doAnswer((i) -> {
@@ -93,6 +103,7 @@ public class ServerTransportFilterTests extends ESTestCase {
                 IndicesOptions.fromOptions(randomBoolean(), randomBoolean(), randomBoolean(), randomBoolean()),
                 randomFrom("*", "_all", "test*"));
         Authentication authentication = mock(Authentication.class);
+        when(authentication.getVersion()).thenReturn(Version.CURRENT);
         when(authentication.getUser()).thenReturn(SystemUser.INSTANCE);
         doAnswer((i) -> {
             ActionListener callback =
@@ -149,6 +160,7 @@ public class ServerTransportFilterTests extends ESTestCase {
             callback.onResponse(empty);
             return Void.TYPE;
         }).when(authzService).roles(any(User.class), any(ActionListener.class));
+        when(authentication.getVersion()).thenReturn(Version.CURRENT);
         when(authentication.getUser()).thenReturn(XPackUser.INSTANCE);
         when(authentication.getRunAsUser()).thenReturn(XPackUser.INSTANCE);
         PlainActionFuture<Void> future = new PlainActionFuture<>();
@@ -163,7 +175,7 @@ public class ServerTransportFilterTests extends ESTestCase {
 
     public void testClientProfileRejectsNodeActions() throws Exception {
         TransportRequest request = mock(TransportRequest.class);
-        ServerTransportFilter filter = getClientFilter();
+        ServerTransportFilter filter = getClientFilter(true);
         ElasticsearchSecurityException e = expectThrows(ElasticsearchSecurityException.class,
                 () -> filter.inbound("internal:foo/bar", request, channel, new PlainActionFuture<>()));
         assertEquals("executing internal/shard actions is considered malicious and forbidden", e.getMessage());
@@ -177,7 +189,7 @@ public class ServerTransportFilterTests extends ESTestCase {
     public void testNodeProfileAllowsNodeActions() throws Exception {
         final String internalAction = "internal:foo/bar";
         final String nodeOrShardAction = "indices:action" + randomFrom("[s]", "[p]", "[r]", "[n]", "[s][p]", "[s][r]", "[f]");
-        ServerTransportFilter filter = getNodeFilter();
+        ServerTransportFilter filter = getNodeFilter(true);
         TransportRequest request = mock(TransportRequest.class);
         Authentication authentication = new Authentication(new User("test", "superuser"), new RealmRef("test", "test", "node1"), null);
         doAnswer((i) -> {
@@ -211,17 +223,65 @@ public class ServerTransportFilterTests extends ESTestCase {
         verifyNoMoreInteractions(authcService, authzService);
     }
 
-    private ServerTransportFilter getClientOrNodeFilter() {
+    public void testHandlesKibanaUserCompatibility() throws Exception {
-        return randomBoolean() ? getNodeFilter() : getClientFilter();
+        TransportRequest request = mock(TransportRequest.class);
+        User user = new User("kibana", "kibana");
+        Authentication authentication = mock(Authentication.class);
+        when(authentication.getVersion())
+                .thenReturn(Version.fromId(randomIntBetween(Version.V_5_0_0_ID, Version.V_5_2_0_ID_UNRELEASED - 100)));
+        when(authentication.getUser()).thenReturn(user);
+        when(authentication.getRunAsUser()).thenReturn(user);
+        doAnswer((i) -> {
+            ActionListener callback =
+                    (ActionListener) i.getArguments()[3];
+            callback.onResponse(authentication);
+            return Void.TYPE;
+        }).when(authcService).authenticate(eq("_action"), eq(request), eq(null), any(ActionListener.class));
+        AtomicReference<String[]> rolesRef = new AtomicReference<>();
+        final Role empty = Role.EMPTY;
+        doAnswer((i) -> {
+            ActionListener callback =
+                    (ActionListener) i.getArguments()[1];
+            rolesRef.set(((User) i.getArguments()[0]).roles());
+            callback.onResponse(empty);
+            return Void.TYPE;
+        }).when(authzService).roles(any(User.class), any(ActionListener.class));
+        ServerTransportFilter filter = getClientOrNodeFilter();
+        PlainActionFuture<Void> future = new PlainActionFuture<>();
+        filter.inbound("_action", request, channel, future);
+        assertNotNull(rolesRef.get());
+        assertThat(rolesRef.get(), arrayContaining("kibana_system"));
+
+        // test with a version that doesn't need changing
+        filter = getClientOrNodeFilter();
+        rolesRef.set(null);
+        user = new KibanaUser(true);
+        when(authentication.getUser()).thenReturn(user);
+        when(authentication.getRunAsUser()).thenReturn(user);
+        when(authentication.getVersion()).thenReturn(Version.V_5_2_0_UNRELEASED);
+        future = new PlainActionFuture<>();
+        filter.inbound("_action", request, channel, future);
+        assertNotNull(rolesRef.get());
+        assertThat(rolesRef.get(), arrayContaining("kibana_system"));
     }
 
-    private ServerTransportFilter.ClientProfile getClientFilter() {
+    private ServerTransportFilter getClientOrNodeFilter() throws IOException {
-        return new ServerTransportFilter.ClientProfile(authcService, authzService, new ThreadContext(Settings.EMPTY), false,
+        return randomBoolean() ? getNodeFilter(true) : getClientFilter(true);
-                destructiveOperations);
     }
 
-    private ServerTransportFilter.NodeProfile getNodeFilter() {
+    private ServerTransportFilter.ClientProfile getClientFilter(boolean reservedRealmEnabled) throws IOException {
-        return new ServerTransportFilter.NodeProfile(authcService, authzService, new ThreadContext(Settings.EMPTY), false,
+        Settings settings = Settings.builder().put("path.home", createTempDir()).build();
-                destructiveOperations);
+        ThreadContext threadContext = new ThreadContext(settings);
+        return new ServerTransportFilter.ClientProfile(authcService, authzService, threadContext, false, destructiveOperations,
+                reservedRealmEnabled,
+                new SecurityContext(settings, threadContext, new CryptoService(Settings.EMPTY, new Environment(settings))));
+    }
+
+    private ServerTransportFilter.NodeProfile getNodeFilter(boolean reservedRealmEnabled) throws IOException {
+        Settings settings = Settings.builder().put("path.home", createTempDir()).build();
+        ThreadContext threadContext = new ThreadContext(settings);
+        return new ServerTransportFilter.NodeProfile(authcService, authzService, threadContext, false, destructiveOperations,
+                reservedRealmEnabled,
+                new SecurityContext(settings, threadContext, new CryptoService(Settings.EMPTY, new Environment(settings))));
     }
 }

elasticsearch/src/test/java/org/elasticsearch/xpack/watcher/test/integration/SearchInputTests.java

@@ -144,7 +144,7 @@ public class SearchInputTests extends ESIntegTestCase {
         XContentParser parser = createParser(builder);
         parser.nextToken();
 
-        SearchRequestParsers searchParsers = new SearchRequestParsers(null, null);
+        SearchRequestParsers searchParsers = new SearchRequestParsers(null);
         SearchInputFactory factory = new SearchInputFactory(Settings.EMPTY, WatcherClientProxy.of(client()),
                                                             searchParsers, xContentRegistry(), scriptService());
 

elasticsearch/src/test/java/org/elasticsearch/xpack/watcher/watch/WatchTests.java

@@ -288,7 +288,7 @@ public class WatchTests extends ESTestCase {
         ActionRegistry actionRegistry = registry(Collections.emptyList(), conditionRegistry, transformRegistry);
         Watch.Parser watchParser = new Watch.Parser(settings, triggerService, actionRegistry, inputRegistry, null, Clock.systemUTC());
 
-        SearchRequestParsers searchParsers = new SearchRequestParsers(null, null);
+        SearchRequestParsers searchParsers = new SearchRequestParsers(null);
         WatcherSearchTemplateService searchTemplateService = new WatcherSearchTemplateService(settings, scriptService, searchParsers,
                 xContentRegistry());
 
@@ -409,7 +409,7 @@ public class WatchTests extends ESTestCase {
         Map<String, InputFactory> parsers = new HashMap<>();
         switch (inputType) {
             case SearchInput.TYPE:
-                SearchRequestParsers searchParsers = new SearchRequestParsers(null, null);
+                SearchRequestParsers searchParsers = new SearchRequestParsers(null);
                 parsers.put(SearchInput.TYPE, new SearchInputFactory(settings, client, searchParsers, xContentRegistry(), scriptService));
                 return new InputRegistry(Settings.EMPTY, parsers);
             default:
@@ -457,7 +457,7 @@ public class WatchTests extends ESTestCase {
     }
 
     private TransformRegistry transformRegistry() {
-        SearchRequestParsers searchParsers = new SearchRequestParsers(null, null);
+        SearchRequestParsers searchParsers = new SearchRequestParsers(null);
         Map<String, TransformFactory> factories = new HashMap<>();
         factories.put(ScriptTransform.TYPE, new ScriptTransformFactory(settings, scriptService));
         factories.put(SearchTransform.TYPE, new SearchTransformFactory(settings, client, searchParsers, xContentRegistry(), scriptService));

elasticsearch/src/test/resources/org/elasticsearch/xpack/security/authz/store/reserved_roles.yml

@@ -23,7 +23,7 @@ superuser:
   run_as:
     - '*'
 
-kibana:
+kibana_system:
   cluster:
     - all
 

qa/rolling-upgrade/build.gradle

@@ -52,12 +52,12 @@ task oldClusterTest(type: RestIntegTestTask) {
     cluster {
         plugin ':x-pack:elasticsearch'
         distribution = 'zip'
-        bwcVersion = '6.0.0-alpha1-SNAPSHOT' // TODO: either randomize, or make this settable with sysprop
+        bwcVersion = '5.3.0-SNAPSHOT' // TODO: either randomize, or make this settable with sysprop
         numBwcNodes = 2
         numNodes = 2
         clusterName = 'rolling-upgrade'
         waitCondition = waitWithAuth
-        systemProperty 'es.logger.org.elasticsearch.xpack.security', 'TRACE'
+        setting 'logger.org.elasticsearch.xpack.security', 'TRACE'
     }
     systemProperty 'tests.rest.suite', 'old_cluster'
 }
@@ -105,7 +105,15 @@ dependencies {
 
 // copy x-pack plugin info so it is on the classpath and security manager has the right permissions
 String outputDir = "generated-resources/${project.name}"
+task copyXPackRestSpec(type: Copy) {
+    dependsOn(project.configurations.restSpec, 'processTestResources')
+    from project(':x-pack:elasticsearch').sourceSets.test.resources
+    include 'rest-api-spec/api/**'
+    into project.sourceSets.test.output.resourcesDir
+}
+
 task copyXPackPluginProps(type: Copy) {
+    dependsOn(copyXPackRestSpec)
     from project(':x-pack:elasticsearch').file('src/main/plugin-metadata')
     from project(':x-pack:elasticsearch').tasks.pluginProperties
     into outputDir

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/30_kibana_write.yml

@@ -0,0 +1,52 @@
+---
+"Verify kibana user role works in mixed cluster":
+  - do:
+      headers:
+        Authorization: "Basic a2liYW5hOmNoYW5nZW1l"
+      cluster.health:
+        wait_for_status: yellow
+        wait_for_nodes: 2
+        timeout: 25s
+  - match: { timed_out: false }
+
+ - do:
+     headers:
+       Authorization: "Basic a2liYW5hOmNoYW5nZW1l"
+     indices.create:
+       index: .kibana-foo
+       wait_for_active_shards : all
+       body:
+         settings:
+           index:
+             number_of_replicas: 1
+
+ - do:
+     headers:
+       Authorization: "Basic a2liYW5hOmNoYW5nZW1l"
+     bulk:
+       refresh: true
+       body:
+         - '{"index": {"_index": ".kibana-foo", "_type": "test_type"}}'
+         - '{"f1": "v1_old", "f2": 0}'
+         - '{"index": {"_index": ".kibana-foo", "_type": "test_type"}}'
+         - '{"f1": "v2_old", "f2": 1}'
+         - '{"index": {"_index": ".kibana-foo", "_type": "test_type"}}'
+         - '{"f1": "v3_old", "f2": 2}'
+         - '{"index": {"_index": ".kibana-foo", "_type": "test_type"}}'
+         - '{"f1": "v4_old", "f2": 3}'
+         - '{"index": {"_index": ".kibana-foo", "_type": "test_type"}}'
+         - '{"f1": "v5_old", "f2": 4}'
+
+ - do:
+     headers:
+       Authorization: "Basic a2liYW5hOmNoYW5nZW1l"
+     indices.flush:
+        index: .kibana-foo
+
+ - do:
+     headers:
+       Authorization: "Basic a2liYW5hOmNoYW5nZW1l"
+     search:
+       index: .kibana-foo
+
+ - match: { hits.total: 5 }