Add ability for plugins to inject roles

src/main/java/com/amazon/opendistroforelasticsearch/security/auth/RolesInjector.java

@@ -0,0 +1,83 @@
+/*
+ *   Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file is distributed
+ *   on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *   express or implied. See the License for the specific language governing
+ *   permissions and limitations under the License.
+ */
+
+package com.amazon.opendistroforelasticsearch.security.auth;
+
+import com.amazon.opendistroforelasticsearch.security.support.ConfigConstants;
+import com.amazon.opendistroforelasticsearch.security.user.User;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.lang.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * This is used to inject opendistro-roles into the request when there is no user involved, like periodic plugin
+ * background jobs. The roles injection is done using thread-context at transport layer only. You can't inject
+ * roles using REST api. Using this we can enforce fine-grained-access-control for the transport layer calls plugins make.
+ *
+ * Format for the injected string: user_name|role_1,role_2
+ * User name is ignored. And roles are opendistro-roles.
+ */
+final public class RolesInjector {
+    protected final Logger log = LogManager.getLogger(RolesInjector.class);
+
+    public RolesInjector() {
+        //empty
+    }
+
+    public Set<String> injectUserAndRoles(final ThreadContext ctx) {
+        final String injectedUserAndRoles = ctx.getTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES);
+        if (injectedUserAndRoles == null) {
+            return null;
+        }
+        log.debug("Injected roles: {}", injectedUserAndRoles);
+
+        String[] strs = injectedUserAndRoles.split("\\|");
+        if (strs.length == 0) {
+            log.error("Roles injected string malformed, could not extract parts. User string was '{}.'" +
+                    " Roles injection failed.", injectedUserAndRoles);
+            return null;
+        }
+
+        if (StringUtils.isEmpty(StringUtils.trim(strs[0]))) {
+            log.error("Username must not be null, injected string was '{}.' Roles injection failed.", injectedUserAndRoles);
+            return null;
+        }
+        User user = new User(strs[0]);
+
+        if (strs.length < 2 || StringUtils.isEmpty(StringUtils.trim(strs[0]))) {
+            log.error("Roles must not be null, injected string was '{}.' Roles injection failed.", injectedUserAndRoles);
+            return null;
+        }
+        Set<String> roles = new HashSet<>(ImmutableSet.copyOf(strs[1].split(",")));
+
+        if(user != null && roles != null) {
+            addUser(user, ctx);
+        }
+        return roles;
+    }
+
+    private void addUser(final User user, final ThreadContext threadContext) {
+        if(threadContext.getTransient(ConfigConstants.OPENDISTRO_SECURITY_USER) != null)
+            return;
+
+        threadContext.putTransient(ConfigConstants.OPENDISTRO_SECURITY_USER, user);
+    }
+}

src/main/java/com/amazon/opendistroforelasticsearch/security/filter/OpenDistroSecurityFilter.java

@@ -35,6 +35,7 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import com.amazon.opendistroforelasticsearch.security.auth.RolesInjector;
 import com.amazon.opendistroforelasticsearch.security.resolver.IndexResolverReplacer;
 import com.amazon.opendistroforelasticsearch.security.support.WildcardMatcher;
 import com.google.common.annotations.VisibleForTesting;
@@ -99,6 +100,7 @@ public class OpenDistroSecurityFilter implements ActionFilter {
     private final CompatConfig compatConfig;
     private final IndexResolverReplacer indexResolverReplacer;
     private final WildcardMatcher immutableIndicesMatcher;
+    private final RolesInjector rolesInjector;
 
     public OpenDistroSecurityFilter(final Settings settings, final PrivilegesEvaluator evalp, final AdminDNs adminDns,
             DlsFlsRequestValve dlsFlsValve, AuditLog auditLog, ThreadPool threadPool, ClusterService cs,
@@ -112,6 +114,7 @@ public OpenDistroSecurityFilter(final Settings settings, final PrivilegesEvaluat
         this.compatConfig = compatConfig;
         this.indexResolverReplacer = indexResolverReplacer;
         this.immutableIndicesMatcher = WildcardMatcher.from(settings.getAsList(ConfigConstants.OPENDISTRO_SECURITY_COMPLIANCE_IMMUTABLE_INDICES, Collections.emptyList()));
+        this.rolesInjector = new RolesInjector();
         log.info("{} indices are made immutable.", immutableIndicesMatcher);
     }
 
@@ -147,7 +150,7 @@ private <Request extends ActionRequest, Response extends ActionResponse> void ap
             if (complianceConfig != null && complianceConfig.isEnabled()) {
                 attachSourceFieldContext(request);
             }
-
+            final Set<String> injectedRoles = rolesInjector.injectUserAndRoles(threadContext);
             final User user = threadContext.getTransient(ConfigConstants.OPENDISTRO_SECURITY_USER);
             final boolean userIsAdmin = isUserAdmin(user, adminDns);
             final boolean interClusterRequest = HeaderHelper.isInterClusterRequest(threadContext);
@@ -229,6 +232,7 @@ private <Request extends ActionRequest, Response extends ActionResponse> void ap
 
             if(Origin.LOCAL.toString().equals(threadContext.getTransient(ConfigConstants.OPENDISTRO_SECURITY_ORIGIN))
                     && (interClusterRequest || HeaderHelper.isDirectRequest(threadContext))
+                    && (injectedRoles == null)
                     ) {
 
                 chain.proceed(task, action, request, listener);
@@ -265,7 +269,7 @@ private <Request extends ActionRequest, Response extends ActionResponse> void ap
                 log.trace("Evaluate permissions for user: {}", user.getName());
             }
 
-            final PrivilegesEvaluatorResponse pres = eval.evaluate(user, action, request, task);
+            final PrivilegesEvaluatorResponse pres = eval.evaluate(user, action, request, task, injectedRoles);
 
             if (log.isDebugEnabled()) {
                 log.debug(pres);
@@ -281,8 +285,11 @@ private <Request extends ActionRequest, Response extends ActionResponse> void ap
                 return;
             } else {
                 auditLog.logMissingPrivileges(action, request, task);
-                log.debug("no permissions for {}", pres.getMissingPrivileges());
-                listener.onFailure(new ElasticsearchSecurityException("no permissions for " + pres.getMissingPrivileges()+ " and "+user, RestStatus.FORBIDDEN));
+                String err = (injectedRoles == null) ?
+                        String.format("no permissions for %s and %s", pres.getMissingPrivileges(), user) :
+                        String.format("no permissions for %s and associated roles %s", pres.getMissingPrivileges(), injectedRoles);
+                log.debug(err);
+                listener.onFailure(new ElasticsearchSecurityException(err, RestStatus.FORBIDDEN));
                 return;
             }
         } catch (Throwable e) {

src/main/java/com/amazon/opendistroforelasticsearch/security/privileges/PrivilegesEvaluator.java

@@ -163,7 +163,8 @@ public boolean isInitialized() {
         return configModel !=null && configModel.getSecurityRoles() != null && dcm != null;
     }
 
-    public PrivilegesEvaluatorResponse evaluate(final User user, String action0, final ActionRequest request, Task task) {
+    public PrivilegesEvaluatorResponse evaluate(final User user, String action0, final ActionRequest request,
+                                                Task task, final Set<String> injectedRoles) {
 
         if (!isInitialized()) {
             throw new ElasticsearchSecurityException("Open Distro Security is not initialized.");
@@ -178,7 +179,7 @@ public PrivilegesEvaluatorResponse evaluate(final User user, String action0, fin
         }
 
         final TransportAddress caller = threadContext.getTransient(ConfigConstants.OPENDISTRO_SECURITY_REMOTE_ADDRESS);
-        final Set<String> mappedRoles = mapRoles(user, caller);
+        final Set<String> mappedRoles = (injectedRoles == null) ? mapRoles(user, caller) : injectedRoles;
         final SecurityRoles securityRoles = getSecurityRoles(mappedRoles);
 
         final PrivilegesEvaluatorResponse presponse = new PrivilegesEvaluatorResponse();
@@ -187,6 +188,7 @@ public PrivilegesEvaluatorResponse evaluate(final User user, String action0, fin
         if (log.isDebugEnabled()) {
             log.debug("### evaluate permissions for {} on {}", user, clusterService.localNode().getName());
             log.debug("action: "+action0+" ("+request.getClass().getSimpleName()+")");
+            log.debug("mapped roles: {}",mappedRoles.toString());
         }
 
         final Resolved requestedResolved = irr.resolveRequest(request);

src/main/java/com/amazon/opendistroforelasticsearch/security/support/ConfigConstants.java

@@ -100,7 +100,7 @@ public class ConfigConstants {
     public static final String OPENDISTRO_SECURITY_USER_HEADER = OPENDISTRO_SECURITY_CONFIG_PREFIX+"user_header";
 
     public static final String OPENDISTRO_SECURITY_INJECTED_USER = "injected_user";
-    
+
     public static final String OPENDISTRO_SECURITY_XFF_DONE = OPENDISTRO_SECURITY_CONFIG_PREFIX+"xff_done";
 
     public static final String SSO_LOGOUT_URL = OPENDISTRO_SECURITY_CONFIG_PREFIX+"sso_logout_url";
@@ -257,6 +257,9 @@ public enum RolesMappingResolution {
     public static final String OPENDISTRO_SECURITY_PROTECTED_INDICES_ROLES_KEY = "opendistro_security.protected_indices.roles";
     public static final List<String> OPENDISTRO_SECURITY_PROTECTED_INDICES_ROLES_DEFAULT = Collections.emptyList();
 
+    // Roles injection for plugins
+    public static final String OPENDISTRO_SECURITY_INJECTED_ROLES = "opendistro_security_injected_roles";
+
     public static Set<String> getSettingAsSet(final Settings settings, final String key, final List<String> defaultList, final boolean ignoreCaseForNone) {
         final List<String> list = settings.getAsList(key, defaultList);
         if (list.size() == 1 && "NONE".equals(ignoreCaseForNone? list.get(0).toUpperCase() : list.get(0))) {

src/test/java/com/amazon/opendistroforelasticsearch/security/RolesInjectorIntegTest.java

@@ -0,0 +1,149 @@
+/*
+ *   Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file is distributed
+ *   on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *   express or implied. See the License for the specific language governing
+ *   permissions and limitations under the License.
+ */
+
+package com.amazon.opendistroforelasticsearch.security;
+
+import com.amazon.opendistroforelasticsearch.security.support.ConfigConstants;
+import com.amazon.opendistroforelasticsearch.security.test.DynamicSecurityConfig;
+import com.amazon.opendistroforelasticsearch.security.test.SingleClusterTest;
+import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.PluginAwareNode;
+import org.elasticsearch.plugins.ActionPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Netty4Plugin;
+import org.elasticsearch.watcher.ResourceWatcherService;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.function.Supplier;
+
+public class RolesInjectorIntegTest extends SingleClusterTest {
+
+    public static class RolesInjectorPlugin extends Plugin implements ActionPlugin {
+        Settings settings;
+        public static String injectedRoles = null;
+
+        public RolesInjectorPlugin(final Settings settings, final Path configPath) {
+            this.settings = settings;
+        }
+
+        @Override
+        public Collection<Object> createComponents(Client client, ClusterService clusterService, ThreadPool threadPool,
+                                                   ResourceWatcherService resourceWatcherService, ScriptService scriptService,
+                                                   NamedXContentRegistry xContentRegistry, Environment environment,
+                                                   NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry,
+                                                   IndexNameExpressionResolver indexNameExpressionResolver,
+                                                   Supplier<RepositoriesService> repositoriesServiceSupplier) {
+            if(injectedRoles != null)
+                threadPool.getThreadContext().putTransient(ConfigConstants.OPENDISTRO_SECURITY_INJECTED_ROLES, injectedRoles);
+            return new ArrayList<>();
+        }
+    }
+
+    //Wait for the security plugin to load roles.
+    private void waitForInit(Client client) throws Exception {
+        try {
+            client.admin().cluster().health(new ClusterHealthRequest()).actionGet();
+        } catch (ElasticsearchSecurityException ex) {
+            if(ex.getMessage().contains("Open Distro Security not initialized")) {
+                Thread.sleep(500);
+                waitForInit(client);
+            }
+        }
+    }
+
+    @Test
+    public void testRolesInject() throws Exception {
+        setup(Settings.EMPTY, new DynamicSecurityConfig().setSecurityRoles("roles.yml"), Settings.EMPTY);
+
+        Assert.assertEquals(clusterInfo.numNodes, clusterHelper.nodeClient().admin().cluster().health(
+                new ClusterHealthRequest().waitForGreenStatus()).actionGet().getNumberOfNodes());
+        Assert.assertEquals(ClusterHealthStatus.GREEN, clusterHelper.nodeClient().admin().cluster().
+                health(new ClusterHealthRequest().waitForGreenStatus()).actionGet().getStatus());
+
+        final Settings tcSettings = Settings.builder()
+                .put(minimumSecuritySettings(Settings.EMPTY).get(0))
+                .put("cluster.name", clusterInfo.clustername)
+                .put("node.data", false)
+                .put("node.master", false)
+                .put("node.ingest", false)
+                .put("path.data", "./target/data/" + clusterInfo.clustername + "/cert/data")
+                .put("path.logs", "./target/data/" + clusterInfo.clustername + "/cert/logs")
+                .put("path.home", "./target")
+                .put("node.name", "testclient")
+                .put("discovery.initial_state_timeout", "8s")
+                .put("opendistro_security.allow_default_init_securityindex", "true")
+                .putList("discovery.zen.ping.unicast.hosts", clusterInfo.nodeHost + ":" + clusterInfo.nodePort)
+                .build();
+
+        //1. Without roles injection.
+        try (Node node = new PluginAwareNode(false, tcSettings, Netty4Plugin.class,
+                OpenDistroSecurityPlugin.class, RolesInjectorPlugin.class).start()) {
+
+            CreateIndexResponse cir = node.client().admin().indices().create(new CreateIndexRequest("captain-logs-1")).actionGet();
+            Assert.assertTrue(cir.isAcknowledged());
+            IndicesExistsResponse ier = node.client().admin().indices().exists(new IndicesExistsRequest("captain-logs-1")).actionGet();
+            Assert.assertTrue(ier.isExists());
+        }
+
+        //2. With invalid roles, must throw security exception.
+        RolesInjectorPlugin.injectedRoles = "invalid_user|invalid_role";
+        Exception exception = null;
+        try (Node node = new PluginAwareNode(false, tcSettings, Netty4Plugin.class, OpenDistroSecurityPlugin.class, RolesInjectorPlugin.class).start()) {
+            waitForInit(node.client());
+
+            CreateIndexResponse cir = node.client().admin().indices().create(new CreateIndexRequest("captain-logs-2")).actionGet();
+            Assert.assertTrue(cir.isAcknowledged());
+        } catch (ElasticsearchSecurityException ex) {
+            exception = ex;
+            log.warn(ex);
+        }
+        Assert.assertNotNull(exception);
+        Assert.assertTrue(exception.getMessage().contains("indices:admin/create"));
+
+        //3. With valid roles - which has permission to create index.
+        RolesInjectorPlugin.injectedRoles = "valid_user|opendistro_security_all_access";
+        try (Node node = new PluginAwareNode(false, tcSettings, Netty4Plugin.class, OpenDistroSecurityPlugin.class, RolesInjectorPlugin.class).start()) {
+            waitForInit(node.client());
+
+            CreateIndexResponse cir = node.client().admin().indices().create(new CreateIndexRequest("captain-logs-3")).actionGet();
+            Assert.assertTrue(cir.isAcknowledged());
+
+            IndicesExistsResponse ier = node.client().admin().indices().exists(new IndicesExistsRequest("captain-logs-3")).actionGet();
+            Assert.assertTrue(ier.isExists());
+        }
+    }
+}