rollback introduction of ServerTransportFilter interface

Author: slobodanadamovic

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/CrossClusterAccessServerTransportFilter.java

@@ -30,7 +30,7 @@
 import static org.elasticsearch.xpack.core.security.authc.CrossClusterAccessSubjectInfo.CROSS_CLUSTER_ACCESS_SUBJECT_INFO_HEADER_KEY;
 import static org.elasticsearch.xpack.security.authc.CrossClusterAccessHeaders.CROSS_CLUSTER_ACCESS_CREDENTIALS_HEADER_KEY;
 
-final class CrossClusterAccessServerTransportFilter extends DefaultServerTransportFilter implements ServerTransportFilter {
+final class CrossClusterAccessServerTransportFilter extends ServerTransportFilter {
 
     private static final Logger logger = LogManager.getLogger(CrossClusterAccessServerTransportFilter.class);
 

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/DefaultServerTransportFilter.java

@@ -106,7 +106,7 @@ private Map<String, ServerTransportFilter> initializeProfileFilters(
             assert profileConfiguration != null : "SSL Profile [" + sslProfile + "] for [" + profileName + "] has a null configuration";
             profileFilters.put(
                 profileName,
-                new DefaultServerTransportFilter(
+                new ServerTransportFilter(
                     authcService,
                     authzService,
                     threadPool.getThreadContext(),

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -4,23 +4,137 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-
 package org.elasticsearch.xpack.security.transport;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.admin.indices.close.TransportCloseIndexAction;
+import org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction;
+import org.elasticsearch.action.admin.indices.open.OpenIndexAction;
+import org.elasticsearch.action.support.DestructiveOperations;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.transport.TaskTransportChannel;
+import org.elasticsearch.transport.TcpChannel;
+import org.elasticsearch.transport.TcpTransportChannel;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.transport.netty4.Netty4TcpChannel;
+import org.elasticsearch.xpack.core.security.SecurityContext;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.user.SystemUser;
+import org.elasticsearch.xpack.security.action.SecurityActionMapper;
+import org.elasticsearch.xpack.security.authc.AuthenticationService;
+import org.elasticsearch.xpack.security.authz.AuthorizationService;
 
 /**
- * Transport filter which is called after a transport request is received by the transport service.
+ * The server transport filter that should be used in nodes as it ensures that an incoming
+ * request is properly authenticated and authorized
  */
-public interface ServerTransportFilter {
+public class ServerTransportFilter {
+
+    private static final Logger logger = LogManager.getLogger(ServerTransportFilter.class);
+
+    private final AuthenticationService authcService;
+    private final AuthorizationService authzService;
+    private final ThreadContext threadContext;
+    private final boolean extractClientCert;
+    private final DestructiveOperations destructiveOperations;
+    private final SecurityContext securityContext;
+
+    public ServerTransportFilter(
+        AuthenticationService authcService,
+        AuthorizationService authzService,
+        ThreadContext threadContext,
+        boolean extractClientCert,
+        DestructiveOperations destructiveOperations,
+        SecurityContext securityContext
+    ) {
+        this.authcService = authcService;
+        this.authzService = authzService;
+        this.threadContext = threadContext;
+        this.extractClientCert = extractClientCert;
+        this.destructiveOperations = destructiveOperations;
+        this.securityContext = securityContext;
+    }
 
     /**
      * Called just after the given request was received by the transport service.
      * <p>
      * Any exception thrown by this method will stop the request from being handled
      * and the error will be sent back to the sender.
      */
-    void inbound(String action, TransportRequest request, TransportChannel transportChannel, ActionListener<Void> listener);
+    public void inbound(String action, TransportRequest request, TransportChannel transportChannel, ActionListener<Void> listener) {
+        if (TransportCloseIndexAction.NAME.equals(action)
+            || OpenIndexAction.NAME.equals(action)
+            || TransportDeleteIndexAction.TYPE.name().equals(action)) {
+            IndicesRequest indicesRequest = (IndicesRequest) request;
+            try {
+                destructiveOperations.failDestructive(indicesRequest.indices());
+            } catch (IllegalArgumentException e) {
+                listener.onFailure(e);
+                return;
+            }
+        }
+        /*
+         here we don't have a fallback user, as all incoming request are
+         expected to have a user attached (either in headers or in context)
+         We can make this assumption because in nodes we make sure all outgoing
+         requests from all the nodes are attached with a user (either a serialize
+         user an authentication token
+         */
+        String securityAction = SecurityActionMapper.action(action, request);
+
+        TransportChannel unwrappedChannel = transportChannel;
+        if (unwrappedChannel instanceof TaskTransportChannel) {
+            unwrappedChannel = ((TaskTransportChannel) unwrappedChannel).getChannel();
+        }
+
+        if (extractClientCert && (unwrappedChannel instanceof TcpTransportChannel)) {
+            TcpChannel tcpChannel = ((TcpTransportChannel) unwrappedChannel).getChannel();
+            if (tcpChannel instanceof Netty4TcpChannel) {
+                if (tcpChannel.isOpen()) {
+                    SSLEngineUtils.extractClientCertificates(logger, threadContext, tcpChannel);
+                }
+            }
+        }
+
+        TransportVersion version = transportChannel.getVersion();
+        authenticate(securityAction, request, listener.delegateFailureAndWrap((l, authentication) -> {
+            if (authentication != null) {
+                if (securityAction.equals(TransportService.HANDSHAKE_ACTION_NAME)
+                    && SystemUser.is(authentication.getEffectiveSubject().getUser()) == false) {
+                    securityContext.executeAsSystemUser(version, original -> {
+                        final Authentication replaced = securityContext.getAuthentication();
+                        authzService.authorize(replaced, securityAction, request, l);
+                    });
+                } else {
+                    authzService.authorize(authentication, securityAction, request, l);
+                }
+            } else {
+                l.onFailure(new IllegalStateException("no authentication present but auth is allowed"));
+            }
+        }));
+    }
+
+    protected void authenticate(
+        final String securityAction,
+        final TransportRequest request,
+        final ActionListener<Authentication> authenticationListener
+    ) {
+        authcService.authenticate(securityAction, request, true, authenticationListener);
+    }
+
+    protected final ThreadContext getThreadContext() {
+        return threadContext;
+    }
+
+    // Package private for testing
+    boolean isExtractClientCert() {
+        return extractClientCert;
+    }
+
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/ServerTransportFilter.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.transport;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.mockito.stubbing.Answer;
+
+import static org.hamcrest.Matchers.arrayWithSize;
+
+public abstract class AbstractServerTransportFilterTests extends ESTestCase {
+
+    protected static Answer<Class<Void>> getAnswer(Authentication authentication) {
+        return getAnswer(authentication, false);
+    }
+
+    protected static Answer<Class<Void>> getAnswer(Authentication authentication, boolean crossClusterAccess) {
+        return i -> {
+            final Object[] args = i.getArguments();
+            assertThat(args, arrayWithSize(crossClusterAccess ? 3 : 4));
+            @SuppressWarnings("unchecked")
+            ActionListener<Authentication> callback = (ActionListener<Authentication>) args[args.length - 1];
+            callback.onResponse(authentication);
+            return Void.TYPE;
+        };
+    }
+
+}

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

@@ -29,7 +29,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-public abstract class ServerTransportInterceptorTests extends ESTestCase {
+public abstract class AbstractServerTransportInterceptorTests extends ESTestCase {
 
     protected static String[] randomRoles() {
         return generateRandomStringArray(3, 10, false, true);