SQL support for field level security (elastic/x-pack-elasticsearch#2162)
This adds support for field level security to SQL by creating a new type of flow for securing requests that look like sql requests. `AuthorizationService` verifies that the user can execute the request but doesn't check the indices in the request because they are not yet ready. Instead, it adds a `BiFunction` to the context that can be used to check permissions for an index while servicing the request. This allows requests to cooperatively secure themselves. SQL does this by implementing filtering on top of its `Catalog` abstraction and backing that filtering with security's filters. This minimizes the touch points between security and SQL. Stuff I'd like to do in followups: What doesn't work at all still: 1. `SHOW TABLES` is still totally unsecured 2. `DESCRIBE TABLE` is still totally unsecured 3. JDBC's metadata APIs are still totally unsecured What kind of works but not well: 1. The audit trail doesn't show the index being authorized for SQL. Original commit: elastic/x-pack-elasticsearch@86f88ba2f5
This commit is contained in:
Nik Everett
GitHub
parent
35389d3be0
commit
f241512e33
|
|
@ -99,6 +99,8 @@ import org.elasticsearch.xpack.security.SecurityFeatureSet;
|
|||
import org.elasticsearch.xpack.security.authc.AuthenticationService;
|
||||
import org.elasticsearch.xpack.security.authc.support.UsernamePasswordToken;
|
||||
import org.elasticsearch.xpack.security.crypto.CryptoService;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.FilteredCatalog;
|
||||
import org.elasticsearch.xpack.sql.plugin.SecurityCatalogFilter;
|
||||
import org.elasticsearch.xpack.sql.plugin.SqlPlugin;
|
||||
import org.elasticsearch.xpack.ssl.SSLConfigurationReloader;
|
||||
import org.elasticsearch.xpack.ssl.SSLService;
|
||||
|
|
@ -324,11 +326,12 @@ public class XPackPlugin extends Plugin implements ScriptPlugin, ActionPlugin, I
|
|||
components.addAll(upgrade.createComponents(internalClient, clusterService, threadPool, resourceWatcherService,
|
||||
scriptService, xContentRegistry));
|
||||
|
||||
FilteredCatalog.Filter securityCatalogFilter = XPackSettings.SECURITY_ENABLED.get(settings) ?
|
||||
new SecurityCatalogFilter(threadPool.getThreadContext(), licenseState) : null;
|
||||
/* Note that we need *client*, not *internalClient* because client preserves the
|
||||
* authenticated user while internalClient throws that user away and acts as the
|
||||
* x-pack user. */
|
||||
components.addAll(sql.createComponents(client, clusterService, threadPool, resourceWatcherService, scriptService, xContentRegistry,
|
||||
environment, nodeEnvironment, namedWriteableRegistry));
|
||||
components.addAll(sql.createComponents(client, clusterService, securityCatalogFilter));
|
||||
|
||||
// just create the reloader as it will pull all of the loaded ssl configurations and start watching them
|
||||
new SSLConfigurationReloader(settings, env, sslService, resourceWatcherService);
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import org.elasticsearch.action.search.ClearScrollAction;
|
|||
import org.elasticsearch.action.search.MultiSearchAction;
|
||||
import org.elasticsearch.action.search.SearchScrollAction;
|
||||
import org.elasticsearch.action.search.SearchTransportService;
|
||||
import org.elasticsearch.action.support.IndicesOptions;
|
||||
import org.elasticsearch.action.support.replication.TransportReplicationAction.ConcreteShardRequest;
|
||||
import org.elasticsearch.action.termvectors.MultiTermVectorsAction;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
|
|
@ -66,16 +67,17 @@ import java.util.Collections;
|
|||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.function.BiFunction;
|
||||
import java.util.function.Predicate;
|
||||
|
||||
import static org.elasticsearch.xpack.security.Security.setting;
|
||||
import static org.elasticsearch.xpack.security.support.Exceptions.authorizationError;
|
||||
|
||||
public class AuthorizationService extends AbstractComponent {
|
||||
|
||||
public static final Setting<Boolean> ANONYMOUS_AUTHORIZATION_EXCEPTION_SETTING =
|
||||
Setting.boolSetting(setting("authc.anonymous.authz_exception"), true, Property.NodeScope);
|
||||
public static final String INDICES_PERMISSIONS_KEY = "_indices_permissions";
|
||||
public static final String INDICES_PERMISSIONS_RESOLVER_KEY = "_indices_permissions_resolver";
|
||||
public static final String ORIGINATING_ACTION_KEY = "_originating_action_name";
|
||||
|
||||
private static final Predicate<String> MONITOR_INDEX_PREDICATE = IndexPrivilege.MONITOR.predicate();
|
||||
|
|
@ -196,6 +198,50 @@ public class AuthorizationService extends AbstractComponent {
|
|||
return;
|
||||
}
|
||||
throw denial(authentication, action, request);
|
||||
} else if (isDelayedIndicesAction(action)) {
|
||||
/* We check now if the user can execute the action without looking at indices.
|
||||
* The action is itself responsible for checking if the user can access the
|
||||
* indices when it runs. */
|
||||
if (permission.indices().check(action)) {
|
||||
grant(authentication, action, request);
|
||||
|
||||
/* Now that we know the user can run the action we need to build a function
|
||||
* that we can use later to fetch the user's actual permissions for an
|
||||
* index. */
|
||||
final MetaData metaData = clusterService.state().metaData();
|
||||
final AuthorizedIndices authorizedIndices = new AuthorizedIndices(authentication.getUser(), permission, action, metaData);
|
||||
|
||||
final TransportRequest finalRequest = request;
|
||||
final Role finalPermission = permission;
|
||||
setIndicesAccessControlResolver((indicesOptions, indices) -> {
|
||||
/* The rest of security assumes that it can extract the indices from a
|
||||
* request and it is fairly twisty to convince it otherwise so we adapt
|
||||
* and stick our indices into a funny proxy request. Not ideal, but
|
||||
* it'll do for now. */
|
||||
IndicesRequest proxy = new IndicesRequest() {
|
||||
@Override
|
||||
public String[] indices() {
|
||||
return indices;
|
||||
}
|
||||
|
||||
@Override
|
||||
public IndicesOptions indicesOptions() {
|
||||
return indicesOptions;
|
||||
}
|
||||
};
|
||||
ResolvedIndices resolvedIndices =
|
||||
resolveIndexNames(authentication, action, proxy, finalRequest, metaData, authorizedIndices);
|
||||
|
||||
Set<String> localIndices = new HashSet<>(resolvedIndices.getLocal());
|
||||
IndicesAccessControl indicesAccessControl =
|
||||
authorizeIndices(action, finalRequest, localIndices, authentication, finalPermission, metaData);
|
||||
// NOCOMMIT we need to rework auditing so we provide the indices
|
||||
grant(authentication, action, finalRequest);
|
||||
return indicesAccessControl;
|
||||
});
|
||||
return;
|
||||
}
|
||||
throw denial(authentication, action, request);
|
||||
} else if (isTranslatedToBulkAction(action)) {
|
||||
if (request instanceof CompositeIndicesRequest == false) {
|
||||
throw new IllegalStateException("Bulk translated actions must implement " + CompositeIndicesRequest.class.getSimpleName()
|
||||
|
|
@ -264,7 +310,7 @@ public class AuthorizationService extends AbstractComponent {
|
|||
|
||||
final MetaData metaData = clusterService.state().metaData();
|
||||
final AuthorizedIndices authorizedIndices = new AuthorizedIndices(authentication.getUser(), permission, action, metaData);
|
||||
final ResolvedIndices resolvedIndices = resolveIndexNames(authentication, action, request, metaData, authorizedIndices);
|
||||
final ResolvedIndices resolvedIndices = resolveIndexNames(authentication, action, request, request, metaData, authorizedIndices);
|
||||
assert !resolvedIndices.isEmpty()
|
||||
: "every indices request needs to have its indices set thus the resolved indices must not be empty";
|
||||
|
||||
|
|
@ -283,22 +329,8 @@ public class AuthorizationService extends AbstractComponent {
|
|||
}
|
||||
|
||||
final Set<String> localIndices = new HashSet<>(resolvedIndices.getLocal());
|
||||
IndicesAccessControl indicesAccessControl = permission.authorize(action, localIndices, metaData, fieldPermissionsCache);
|
||||
if (!indicesAccessControl.isGranted()) {
|
||||
throw denial(authentication, action, request);
|
||||
} else if (indicesAccessControl.getIndexPermissions(SecurityLifecycleService.SECURITY_INDEX_NAME) != null
|
||||
&& indicesAccessControl.getIndexPermissions(SecurityLifecycleService.SECURITY_INDEX_NAME).isGranted()
|
||||
&& XPackUser.is(authentication.getUser()) == false
|
||||
&& MONITOR_INDEX_PREDICATE.test(action) == false
|
||||
&& isSuperuser(authentication.getUser()) == false) {
|
||||
// only the XPackUser is allowed to work with this index, but we should allow indices monitoring actions through for debugging
|
||||
// purposes. These monitor requests also sometimes resolve indices concretely and then requests them
|
||||
logger.debug("user [{}] attempted to directly perform [{}] against the security index [{}]",
|
||||
authentication.getUser().principal(), action, SecurityLifecycleService.SECURITY_INDEX_NAME);
|
||||
throw denial(authentication, action, request);
|
||||
} else {
|
||||
setIndicesAccessControl(indicesAccessControl);
|
||||
}
|
||||
IndicesAccessControl indicesAccessControl = authorizeIndices(action, request, localIndices, authentication, permission, metaData);
|
||||
setIndicesAccessControl(indicesAccessControl);
|
||||
|
||||
//if we are creating an index we need to authorize potential aliases created at the same time
|
||||
if (IndexPrivilege.CREATE_INDEX_MATCHER.test(action)) {
|
||||
|
|
@ -321,20 +353,27 @@ public class AuthorizationService extends AbstractComponent {
|
|||
grant(authentication, action, originalRequest);
|
||||
}
|
||||
|
||||
private ResolvedIndices resolveIndexNames(Authentication authentication, String action, TransportRequest request, MetaData metaData,
|
||||
AuthorizedIndices authorizedIndices) {
|
||||
private ResolvedIndices resolveIndexNames(Authentication authentication, String action, Object indicesRequest,
|
||||
TransportRequest mainRequest, MetaData metaData, AuthorizedIndices authorizedIndices) {
|
||||
try {
|
||||
return indicesAndAliasesResolver.resolve(request, metaData, authorizedIndices);
|
||||
return indicesAndAliasesResolver.resolve(indicesRequest, metaData, authorizedIndices);
|
||||
} catch (Exception e) {
|
||||
auditTrail.accessDenied(authentication.getUser(), action, request);
|
||||
auditTrail.accessDenied(authentication.getUser(), action, mainRequest);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
private void setIndicesAccessControl(IndicesAccessControl accessControl) {
|
||||
if (threadContext.getTransient(INDICES_PERMISSIONS_KEY) == null) {
|
||||
threadContext.putTransient(INDICES_PERMISSIONS_KEY, accessControl);
|
||||
}
|
||||
putTransientIfNonExisting(INDICES_PERMISSIONS_KEY, accessControl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets a function to resolve {@link IndicesAccessControl} to be used by
|
||||
* {@link #isDelayedIndicesAction(String) actions} that do not know their
|
||||
* indices up front but still need to check permissions.
|
||||
*/
|
||||
private void setIndicesAccessControlResolver(BiFunction<IndicesOptions, String[], IndicesAccessControl> accessControlResolver) {
|
||||
putTransientIfNonExisting(INDICES_PERMISSIONS_RESOLVER_KEY, accessControlResolver);
|
||||
}
|
||||
|
||||
private void putTransientIfNonExisting(String key, Object value) {
|
||||
|
|
@ -385,8 +424,17 @@ public class AuthorizationService extends AbstractComponent {
|
|||
action.equals("indices:data/read/mpercolate") ||
|
||||
action.equals("indices:data/read/msearch/template") ||
|
||||
action.equals("indices:data/read/search/template") ||
|
||||
action.equals("indices:data/write/reindex") ||
|
||||
action.equals(SqlAction.NAME) || // NOCOMMIT verify that SQL requests are properly composite
|
||||
action.equals("indices:data/write/reindex");
|
||||
}
|
||||
|
||||
/**
|
||||
* Delayed actions are authorized at start time but do not know which
|
||||
* indices they target when the start so {@link AuthorizationService}
|
||||
* sets up a function that can be used to authorize indices during
|
||||
* the request.
|
||||
*/
|
||||
private static boolean isDelayedIndicesAction(String action) {
|
||||
return action.equals(SqlAction.NAME) ||
|
||||
action.equals(JdbcAction.NAME) ||
|
||||
action.equals(CliAction.NAME);
|
||||
}
|
||||
|
|
@ -410,6 +458,30 @@ public class AuthorizationService extends AbstractComponent {
|
|||
action.equals(SearchTransportService.CLEAR_SCROLL_CONTEXTS_ACTION_NAME);
|
||||
}
|
||||
|
||||
/**
|
||||
* Authorize some indices, throwing an exception if they are not authorized and returning
|
||||
* the {@link IndicesAccessControl} if they are.
|
||||
*/
|
||||
private IndicesAccessControl authorizeIndices(String action, TransportRequest request, Set<String> localIndices,
|
||||
Authentication authentication, Role permission, MetaData metaData) {
|
||||
IndicesAccessControl indicesAccessControl = permission.authorize(action, localIndices, metaData, fieldPermissionsCache);
|
||||
if (!indicesAccessControl.isGranted()) {
|
||||
throw denial(authentication, action, request);
|
||||
}
|
||||
if (indicesAccessControl.getIndexPermissions(SecurityLifecycleService.SECURITY_INDEX_NAME) != null
|
||||
&& indicesAccessControl.getIndexPermissions(SecurityLifecycleService.SECURITY_INDEX_NAME).isGranted()
|
||||
&& XPackUser.is(authentication.getUser()) == false
|
||||
&& MONITOR_INDEX_PREDICATE.test(action) == false
|
||||
&& isSuperuser(authentication.getUser()) == false) {
|
||||
// only the superusers are allowed to work with this index, but we should allow indices monitoring actions through for debugging
|
||||
// purposes. These monitor requests also sometimes resolve indices concretely and then requests them
|
||||
logger.debug("user [{}] attempted to directly perform [{}] against the security index [{}]",
|
||||
authentication.getUser().principal(), action, SecurityLifecycleService.SECURITY_INDEX_NAME);
|
||||
throw denial(authentication, action, request);
|
||||
}
|
||||
return indicesAccessControl;
|
||||
}
|
||||
|
||||
static boolean checkSameUserPermissions(String action, TransportRequest request, Authentication authentication) {
|
||||
final boolean actionAllowed = SAME_USER_PRIVILEGE.test(action);
|
||||
if (actionAllowed) {
|
||||
|
|
|
|||
|
|
@ -5,18 +5,6 @@
|
|||
*/
|
||||
package org.elasticsearch.xpack.security.authz;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.SortedMap;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.elasticsearch.action.AliasesRequest;
|
||||
import org.elasticsearch.action.IndicesRequest;
|
||||
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
|
||||
|
|
@ -36,9 +24,20 @@ import org.elasticsearch.common.settings.ClusterSettings;
|
|||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.index.IndexNotFoundException;
|
||||
import org.elasticsearch.transport.RemoteClusterAware;
|
||||
import org.elasticsearch.transport.TransportRequest;
|
||||
import org.elasticsearch.xpack.graph.action.GraphExploreRequest;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.SortedMap;
|
||||
import java.util.concurrent.CopyOnWriteArraySet;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
public class IndicesAndAliasesResolver {
|
||||
|
||||
//placeholder used in the security plugin to indicate that the request is authorized knowing that it will yield an empty response
|
||||
|
|
@ -84,8 +83,7 @@ public class IndicesAndAliasesResolver {
|
|||
* <br>
|
||||
* Otherwise, <em>N</em> will be added to the <em>local</em> index list.
|
||||
*/
|
||||
|
||||
public ResolvedIndices resolve(TransportRequest request, MetaData metaData, AuthorizedIndices authorizedIndices) {
|
||||
public ResolvedIndices resolve(Object request, MetaData metaData, AuthorizedIndices authorizedIndices) {
|
||||
if (request instanceof IndicesAliasesRequest) {
|
||||
ResolvedIndices indices = ResolvedIndices.empty();
|
||||
IndicesAliasesRequest indicesAliasesRequest = (IndicesAliasesRequest) request;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,78 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.plugin;
|
||||
|
||||
import org.elasticsearch.action.support.IndicesOptions;
|
||||
import org.elasticsearch.common.util.concurrent.ThreadContext;
|
||||
import org.elasticsearch.license.XPackLicenseState;
|
||||
import org.elasticsearch.xpack.security.authz.AuthorizationService;
|
||||
import org.elasticsearch.xpack.security.authz.accesscontrol.IndicesAccessControl;
|
||||
import org.elasticsearch.xpack.security.authz.accesscontrol.IndicesAccessControl.IndexAccessControl;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsIndex;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.FilteredCatalog;
|
||||
import org.elasticsearch.xpack.sql.type.DataType;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
/**
|
||||
* Adapts SQL to x-pack security by intercepting calls to its catalog and
|
||||
* authorizing the user's view of every index, filtering out fields that
|
||||
* the user does not have access to.
|
||||
* <p>
|
||||
* Document level security is handled using the standard search mechanisms
|
||||
* but this class is required for SQL to respect field level security for
|
||||
* {@code SELECT} statements and index level security for metadata
|
||||
* statements like {@code SHOW TABLES} and {@code DESCRIBE TABLE}.
|
||||
*/
|
||||
public class SecurityCatalogFilter implements FilteredCatalog.Filter {
|
||||
// NOCOMMIT need to figure out sql on aliases that expand to many indices
|
||||
private static final IndicesOptions OPTIONS = IndicesOptions.strictSingleIndexNoExpandForbidClosed();
|
||||
|
||||
private final ThreadContext threadContext;
|
||||
private final XPackLicenseState licenseState;
|
||||
|
||||
public SecurityCatalogFilter(ThreadContext threadContext, XPackLicenseState licenseState) {
|
||||
this.threadContext = threadContext;
|
||||
this.licenseState = licenseState;
|
||||
}
|
||||
|
||||
@Override
|
||||
public EsIndex filterIndex(EsIndex index) {
|
||||
if (false == licenseState.isAuthAllowed()) {
|
||||
/* If security is disabled the index authorization won't be available.
|
||||
* It is technically possible there to be a race between licensing
|
||||
* being enabled and sql requests that might cause us to fail on those
|
||||
* requests but that should be rare. */
|
||||
return index;
|
||||
}
|
||||
IndexAccessControl permissions = getAccessControlResolver()
|
||||
.apply(OPTIONS, new String[] {index.name()})
|
||||
.getIndexPermissions(index.name());
|
||||
/* Fetching the permissions always checks if they are granted. If they aren't
|
||||
* then it throws an exception. */
|
||||
if (false == permissions.getFieldPermissions().hasFieldLevelSecurity()) {
|
||||
return index;
|
||||
}
|
||||
Map<String, DataType> filteredMapping = new HashMap<>(index.mapping().size());
|
||||
for (Map.Entry<String, DataType> entry : index.mapping().entrySet()) {
|
||||
if (permissions.getFieldPermissions().grantsAccessTo(entry.getKey())) {
|
||||
filteredMapping.put(entry.getKey(), entry.getValue());
|
||||
}
|
||||
}
|
||||
return new EsIndex(index.name(), filteredMapping, index.aliases(), index.settings());
|
||||
}
|
||||
|
||||
private BiFunction<IndicesOptions, String[], IndicesAccessControl> getAccessControlResolver() {
|
||||
BiFunction<IndicesOptions, String[], IndicesAccessControl> resolver =
|
||||
threadContext.getTransient(AuthorizationService.INDICES_PERMISSIONS_RESOLVER_KEY);
|
||||
if (resolver == null) {
|
||||
throw new IllegalStateException("SQL request wasn't recognized properly by security");
|
||||
}
|
||||
return resolver;
|
||||
}
|
||||
}
|
||||
|
|
@ -5,14 +5,6 @@
|
|||
*/
|
||||
package org.elasticsearch.xpack.security.authz;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.elasticsearch.ElasticsearchSecurityException;
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
|
|
@ -50,8 +42,6 @@ import org.elasticsearch.action.admin.indices.upgrade.get.UpgradeStatusAction;
|
|||
import org.elasticsearch.action.admin.indices.upgrade.get.UpgradeStatusRequest;
|
||||
import org.elasticsearch.action.bulk.BulkAction;
|
||||
import org.elasticsearch.action.bulk.BulkRequest;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.common.settings.ClusterSettings;
|
||||
import org.elasticsearch.action.delete.DeleteAction;
|
||||
import org.elasticsearch.action.delete.DeleteRequest;
|
||||
import org.elasticsearch.action.get.GetAction;
|
||||
|
|
@ -81,12 +71,14 @@ import org.elasticsearch.client.Client;
|
|||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.service.ClusterService;
|
||||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.collect.Tuple;
|
||||
import org.elasticsearch.common.settings.ClusterSettings;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.util.concurrent.ThreadContext;
|
||||
import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
|
||||
import org.elasticsearch.index.IndexNotFoundException;
|
||||
import org.elasticsearch.license.GetLicenseAction;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
|
|
@ -123,8 +115,18 @@ import org.elasticsearch.xpack.security.user.ElasticUser;
|
|||
import org.elasticsearch.xpack.security.user.SystemUser;
|
||||
import org.elasticsearch.xpack.security.user.User;
|
||||
import org.elasticsearch.xpack.security.user.XPackUser;
|
||||
import org.elasticsearch.xpack.sql.plugin.sql.action.SqlAction;
|
||||
import org.junit.Before;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
import static org.elasticsearch.test.SecurityTestsUtils.assertAuthenticationException;
|
||||
import static org.elasticsearch.test.SecurityTestsUtils.assertThrowsAuthorizationException;
|
||||
import static org.elasticsearch.test.SecurityTestsUtils.assertThrowsAuthorizationExceptionRunAs;
|
||||
|
|
@ -890,6 +892,67 @@ public class AuthorizationServiceTests extends ESTestCase {
|
|||
() -> authorize(createAuthentication(userDenied), action, request), action, "userDenied");
|
||||
}
|
||||
|
||||
public void testDelayedActionsAddResolver() {
|
||||
String action = SqlAction.NAME;
|
||||
MockIndicesRequest request = new MockIndicesRequest(IndicesOptions.strictExpandOpen(), "index");
|
||||
|
||||
User userAllowed = new User("userAllowed", "roleAllowed");
|
||||
roleMap.put("roleAllowed", new RoleDescriptor("roleAllowed", null,
|
||||
new IndicesPrivileges[] { IndicesPrivileges.builder().indices("index").privileges("all").build() }, null));
|
||||
User userDenied = new User("userDenied", "roleDenied");
|
||||
roleMap.put("roleDenied", new RoleDescriptor("roleDenied", null,
|
||||
new IndicesPrivileges[] { IndicesPrivileges.builder().indices("a").privileges("all").build() }, null));
|
||||
User userSome = new User("userSome", "roleSome");
|
||||
roleMap.put("roleSome", new RoleDescriptor("roleSome", null,
|
||||
new IndicesPrivileges[] {
|
||||
IndicesPrivileges.builder().indices("a").privileges("all").build(),
|
||||
IndicesPrivileges.builder().indices("b").privileges("all").build()
|
||||
}, null));
|
||||
mockEmptyMetaData();
|
||||
assertNull(getAccessControlResolver());
|
||||
|
||||
try (StoredContext ctxRestore = threadContext.stashContext()) {
|
||||
authorize(createAuthentication(userAllowed), action, request);
|
||||
// NOCOMMIT we need to test the audit trail but it needs to be improved to allow explicit indices.
|
||||
assertNotNull(getAccessControlResolver());
|
||||
IndicesAccessControl iac = getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"index"});
|
||||
assertTrue(iac.isGranted());
|
||||
assertTrue(iac.getIndexPermissions("index").isGranted());
|
||||
}
|
||||
assertNull(getAccessControlResolver());
|
||||
|
||||
try (StoredContext ctxRestore = threadContext.stashContext()) {
|
||||
authorize(createAuthentication(userDenied), action, request);
|
||||
assertNotNull(getAccessControlResolver());
|
||||
assertThrowsAuthorizationException(
|
||||
() -> getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"index"}),
|
||||
action, "userDenied");
|
||||
}
|
||||
assertNull(getAccessControlResolver());
|
||||
|
||||
try (StoredContext ctxRestore = threadContext.stashContext()) {
|
||||
authorize(createAuthentication(userSome), action, request);
|
||||
assertNotNull(getAccessControlResolver());
|
||||
IndicesAccessControl iac = getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"a"});
|
||||
assertTrue(iac.isGranted());
|
||||
assertTrue(iac.getIndexPermissions("a").isGranted());
|
||||
|
||||
iac = getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"b"});
|
||||
assertTrue(iac.isGranted());
|
||||
assertTrue(iac.getIndexPermissions("b").isGranted());
|
||||
|
||||
iac = getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"a", "b"});
|
||||
assertTrue(iac.isGranted());
|
||||
assertTrue(iac.getIndexPermissions("a").isGranted());
|
||||
assertTrue(iac.getIndexPermissions("b").isGranted());
|
||||
|
||||
assertThrowsAuthorizationException(
|
||||
() -> getAccessControlResolver().apply(IndicesOptions.strictExpandOpen(), new String[] {"index", "a", "b"}),
|
||||
action, "userSome");
|
||||
}
|
||||
assertNull(getAccessControlResolver());
|
||||
}
|
||||
|
||||
public void testSameUserPermission() {
|
||||
final User user = new User("joe");
|
||||
final boolean changePasswordRequest = randomBoolean();
|
||||
|
|
@ -1155,4 +1218,8 @@ public class AuthorizationServiceTests extends ESTestCase {
|
|||
() -> authorize(createAuthentication(user), action, transportRequest), action, "test user");
|
||||
verify(auditTrail).accessDenied(user, action, clearScrollRequest);
|
||||
}
|
||||
|
||||
private BiFunction<IndicesOptions, String[], IndicesAccessControl> getAccessControlResolver() {
|
||||
return threadContext.getTransient(AuthorizationService.INDICES_PERMISSIONS_RESOLVER_KEY);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,9 +14,9 @@ import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
|||
public class TestUtils {
|
||||
// NOCOMMIT I think these may not be needed if we switch to integration tests for the protos
|
||||
public static PlanExecutor planExecutor(Client client) {
|
||||
PlanExecutor executor = new PlanExecutor(client,
|
||||
() -> client.admin().cluster().prepareState().get(TimeValue.timeValueMinutes(1)).getState());
|
||||
((EsCatalog) executor.catalog()).setIndexNameExpressionResolver(new IndexNameExpressionResolver(client.settings()));
|
||||
EsCatalog catalog = new EsCatalog(() -> client.admin().cluster().prepareState().get(TimeValue.timeValueMinutes(1)).getState());
|
||||
catalog.setIndexNameExpressionResolver(new IndexNameExpressionResolver(client.settings()));
|
||||
PlanExecutor executor = new PlanExecutor(client, catalog);
|
||||
return executor;
|
||||
}
|
||||
}
|
||||
|
|
@ -27,6 +27,10 @@ import static java.util.Collections.singletonMap;
|
|||
import static org.hamcrest.Matchers.both;
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
|
||||
/**
|
||||
* Integration test for the rest sql action. That one that speaks json directly to a
|
||||
* user rather than to the JDBC driver or CLI.
|
||||
*/
|
||||
public class RestSqlIT extends ESRestTestCase {
|
||||
public void testBasicQuery() throws IOException {
|
||||
StringBuilder bulk = new StringBuilder();
|
||||
|
|
@ -64,6 +68,10 @@ public class RestSqlIT extends ESRestTestCase {
|
|||
new StringEntity("{\"query\":\"SELECT DAY_OF_YEAR(test), COUNT(*) FROM test.test\"}", ContentType.APPLICATION_JSON)));
|
||||
}
|
||||
|
||||
public void testMissingIndex() throws IOException {
|
||||
expectBadRequest(() -> runSql("SELECT foo FROM missing"), containsString("1:17: Cannot resolve index [missing]"));
|
||||
}
|
||||
|
||||
public void testMissingField() throws IOException {
|
||||
StringBuilder bulk = new StringBuilder();
|
||||
bulk.append("{\"index\":{\"_id\":\"1\"}}\n");
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ import org.elasticsearch.common.xcontent.XContentHelper;
|
|||
import org.elasticsearch.common.xcontent.json.JsonXContent;
|
||||
import org.elasticsearch.test.NotEqualMessageBuilder;
|
||||
import org.elasticsearch.test.rest.ESRestTestCase;
|
||||
import org.hamcrest.Matcher;
|
||||
import org.junit.Before;
|
||||
|
||||
import java.io.IOException;
|
||||
|
|
@ -60,6 +61,7 @@ public class SqlSecurityIT extends ESRestTestCase {
|
|||
// NOCOMMIT we're going to need to test jdbc and cli with these too!
|
||||
// NOCOMMIT we'll have to test scrolling as well
|
||||
// NOCOMMIT tests for describing a table and showing tables
|
||||
// NOCOMMIT tests with audit trail
|
||||
|
||||
public void testSqlWorksAsAdmin() throws IOException {
|
||||
Map<String, Object> expected = new HashMap<>();
|
||||
|
|
@ -101,28 +103,18 @@ public class SqlSecurityIT extends ESRestTestCase {
|
|||
assertThat(e.getMessage(), containsString("403 Forbidden"));
|
||||
}
|
||||
|
||||
@AwaitsFix(bugUrl="https://github.com/elastic/x-pack-elasticsearch/issues/2074")
|
||||
public void testSqlSingleFieldGranted() throws IOException {
|
||||
createUser("only_a", "read_test_a");
|
||||
|
||||
/* This doesn't work because sql doesn't see the field level security
|
||||
* and still adds the metadata even though the field "doesn't exist" */
|
||||
assertResponse(runSql("SELECT a FROM test.test", null), runSql("SELECT * FROM test", "only_a"));
|
||||
/* This should probably be a 400 level error complaining about field
|
||||
* that do not exist because that is what makes sense in SQL. */
|
||||
assertResponse(emptyMap(), runSql("SELECT * FROM test.test WHERE c = 3", "only_a"));
|
||||
assertResponse(runSql("SELECT a FROM test", null), runSql("SELECT * FROM test", "only_a"));
|
||||
expectBadRequest(() -> runSql("SELECT c FROM test", "only_a"), containsString("line 1:8: Unresolved item 'c'"));
|
||||
}
|
||||
|
||||
@AwaitsFix(bugUrl="https://github.com/elastic/x-pack-elasticsearch/issues/2074")
|
||||
public void testSqlSingleFieldExcepted() throws IOException {
|
||||
createUser("not_c", "read_test_a_and_b");
|
||||
|
||||
/* This doesn't work because sql doesn't see the field level security
|
||||
* and still adds the metadata even though the field "doesn't exist" */
|
||||
assertResponse(runSql("SELECT a, b FROM test", null), runSql("SELECT * FROM test", "not_c"));
|
||||
/* This should probably be a 400 level error complaining about field
|
||||
* that do not exist because that is what makes sense in SQL. */
|
||||
assertResponse(emptyMap(), runSql("SELECT * FROM test WHERE c = 3", "not_c"));
|
||||
expectBadRequest(() -> runSql("SELECT c FROM test", "not_c"), containsString("line 1:8: Unresolved item 'c'"));
|
||||
}
|
||||
|
||||
public void testSqlDocumentExclued() throws IOException {
|
||||
|
|
@ -131,6 +123,12 @@ public class SqlSecurityIT extends ESRestTestCase {
|
|||
assertResponse(runSql("SELECT * FROM test WHERE c != 3", null), runSql("SELECT * FROM test", "no_3s"));
|
||||
}
|
||||
|
||||
private void expectBadRequest(ThrowingRunnable code, Matcher<String> errorMessageMatcher) {
|
||||
ResponseException e = expectThrows(ResponseException.class, code);
|
||||
assertEquals(400, e.getResponse().getStatusLine().getStatusCode());
|
||||
assertThat(e.getMessage(), errorMessageMatcher);
|
||||
}
|
||||
|
||||
private void assertResponse(Map<String, Object> expected, Map<String, Object> actual) {
|
||||
if (false == expected.equals(actual)) {
|
||||
NotEqualMessageBuilder message = new NotEqualMessageBuilder();
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ public class ErrorsIT extends JdbcIntegrationTestCase {
|
|||
public void testSelectFromMissingTable() throws Exception {
|
||||
try (Connection c = esJdbc()) {
|
||||
SQLException e = expectThrows(SQLException.class, () -> c.prepareStatement("SELECT * from test").executeQuery());
|
||||
assertEquals("line 1:15: Cannot resolve index test", e.getMessage());
|
||||
assertEquals("line 1:15: Cannot resolve index [test]", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,6 @@ import org.elasticsearch.xpack.sql.tree.Node;
|
|||
public class UnknownIndexException extends AnalysisException {
|
||||
|
||||
public UnknownIndexException(String index, Node<?> source) {
|
||||
super(source, "Cannot resolve index %s", index);
|
||||
super(source, "Cannot resolve index [%s]", index);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import org.elasticsearch.xpack.sql.analysis.UnknownFunctionException;
|
|||
import org.elasticsearch.xpack.sql.analysis.UnknownIndexException;
|
||||
import org.elasticsearch.xpack.sql.analysis.analyzer.Verifier.Failure;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.Catalog;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsIndex;
|
||||
import org.elasticsearch.xpack.sql.capabilities.Resolvables;
|
||||
import org.elasticsearch.xpack.sql.expression.Alias;
|
||||
import org.elasticsearch.xpack.sql.expression.Attribute;
|
||||
|
|
@ -237,21 +238,19 @@ public class Analyzer extends RuleExecutor<LogicalPlan> {
|
|||
}
|
||||
|
||||
private class ResolveTable extends AnalyzeRule<UnresolvedRelation> {
|
||||
|
||||
@Override
|
||||
protected LogicalPlan rule(UnresolvedRelation plan) {
|
||||
TableIdentifier table = plan.table();
|
||||
String index = table.index();
|
||||
if (!catalog.indexExists(index)) {
|
||||
throw new UnknownIndexException(index, plan);
|
||||
if (!catalog.indexIsValid(table.index())) {
|
||||
throw new InvalidIndexException(table.index(), plan);
|
||||
}
|
||||
|
||||
if (!catalog.indexIsValid(index)) {
|
||||
throw new InvalidIndexException(index, plan);
|
||||
EsIndex found = catalog.getIndex(table.index());
|
||||
if (found == null) {
|
||||
throw new UnknownIndexException(table.index(), plan);
|
||||
}
|
||||
|
||||
LogicalPlan catalogTable = new CatalogTable(plan.location(), catalog.getIndex(index));
|
||||
SubQueryAlias sa = new SubQueryAlias(plan.location(), catalogTable, index);
|
||||
LogicalPlan catalogTable = new CatalogTable(plan.location(), found);
|
||||
SubQueryAlias sa = new SubQueryAlias(plan.location(), catalogTable, table.index());
|
||||
|
||||
if (plan.alias() != null) {
|
||||
sa = new SubQueryAlias(plan.location(), sa, plan.alias());
|
||||
|
|
|
|||
|
|
@ -5,18 +5,24 @@
|
|||
*/
|
||||
package org.elasticsearch.xpack.sql.analysis.catalog;
|
||||
|
||||
import org.elasticsearch.common.Nullable;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
|
||||
public interface Catalog {
|
||||
/**
|
||||
* Check if an index is valid for sql.
|
||||
*/
|
||||
boolean indexIsValid(String index); // NOCOMMIT should probably be merged into EsCatalog's getIndex method.
|
||||
|
||||
boolean indexExists(String index);
|
||||
|
||||
boolean indexIsValid(String index);
|
||||
|
||||
/**
|
||||
* Lookup the information for a table, returning {@code null} if
|
||||
* the index is not found.
|
||||
*/
|
||||
@Nullable
|
||||
EsIndex getIndex(String index);
|
||||
|
||||
List<EsIndex> listIndices();
|
||||
|
||||
List<EsIndex> listIndices(String pattern);
|
||||
// NOCOMMIT should these be renamed to getTable and listTables? That seems like a name given that this is a SQL Catalog abstraction.
|
||||
}
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import org.elasticsearch.cluster.ClusterState;
|
|||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
|
||||
import org.elasticsearch.cluster.metadata.MetaData;
|
||||
import org.elasticsearch.common.inject.Inject;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
|
|
@ -25,7 +26,7 @@ public class EsCatalog implements Catalog {
|
|||
this.clusterState = clusterState;
|
||||
}
|
||||
|
||||
// initialization hack
|
||||
@Inject // NOCOMMIT more to ctor and move resolver to createComponents
|
||||
public void setIndexNameExpressionResolver(IndexNameExpressionResolver resolver) {
|
||||
this.indexNameExpressionResolver = resolver;
|
||||
}
|
||||
|
|
@ -36,27 +37,18 @@ public class EsCatalog implements Catalog {
|
|||
|
||||
@Override
|
||||
public EsIndex getIndex(String index) {
|
||||
if (!indexExists(index)) {
|
||||
return EsIndex.NOT_FOUND;
|
||||
MetaData metadata = metadata();
|
||||
if (false == metadata.hasIndex(index)) {
|
||||
return null;
|
||||
}
|
||||
return EsIndex.build(metadata().index(index));
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean indexExists(String index) {
|
||||
IndexMetaData idx = metadata().index(index);
|
||||
return idx != null;
|
||||
return EsIndex.build(metadata.index(index));
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean indexIsValid(String index) {
|
||||
// NOCOMMIT there is a race condition here with indices being deleted. This should be moved into getIndex
|
||||
IndexMetaData idx = metadata().index(index);
|
||||
return idx != null && indexHasOnlyOneType(idx);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<EsIndex> listIndices() {
|
||||
return listIndices(null);
|
||||
return idx == null || indexHasOnlyOneType(idx);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -92,6 +84,7 @@ public class EsCatalog implements Catalog {
|
|||
}
|
||||
|
||||
private String[] resolveIndex(String pattern) {
|
||||
// NOCOMMIT we should use the cluster state that we resolve when we fetch the metadata so it is the *same* so we don't have weird errors when indices are deleted
|
||||
return indexNameExpressionResolver.concreteIndexNames(clusterState.get(), IndicesOptions.strictExpandOpenAndForbidClosed(), pattern);
|
||||
}
|
||||
}
|
||||
|
|
@ -27,6 +27,7 @@ public class EsIndex {
|
|||
|
||||
public static final EsIndex NOT_FOUND = new EsIndex(StringUtils.EMPTY, emptyMap(), emptyList(), Settings.EMPTY);
|
||||
|
||||
// NOCOMMIT Double check that we need these and that we're securing them properly.
|
||||
private final String name;
|
||||
private final Map<String, DataType> mapping;
|
||||
private final List<String> aliases;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,51 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.analysis.catalog;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* {@link Catalog} implementation that filters the results.
|
||||
*/
|
||||
public class FilteredCatalog implements Catalog {
|
||||
public interface Filter {
|
||||
/**
|
||||
* Filter an index. Returning {@code null} will act as though
|
||||
* the index wasn't found. Will never be called with a {@code null}
|
||||
* parameter.
|
||||
*/
|
||||
EsIndex filterIndex(EsIndex index);
|
||||
}
|
||||
|
||||
private Catalog delegate;
|
||||
private Filter filter;
|
||||
|
||||
public FilteredCatalog(Catalog delegate, Filter filter) {
|
||||
this.delegate = delegate;
|
||||
this.filter = filter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<EsIndex> listIndices(String pattern) {
|
||||
// NOCOMMIT authorize me
|
||||
return delegate.listIndices(pattern);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean indexIsValid(String index) {
|
||||
return delegate.indexIsValid(index);
|
||||
}
|
||||
|
||||
@Override
|
||||
public EsIndex getIndex(String index) {
|
||||
// NOCOMMIT we need to think really carefully about how we deal with aliases that resolve into multiple indices.
|
||||
EsIndex result = delegate.getIndex(index);
|
||||
if (result == null) {
|
||||
return null;
|
||||
}
|
||||
return filter.filterIndex(result);
|
||||
}
|
||||
}
|
||||
|
|
@ -36,11 +36,11 @@ public class PlanExecutor extends AbstractLifecycleComponent {
|
|||
private final Optimizer optimizer;
|
||||
private final Planner planner;
|
||||
|
||||
public PlanExecutor(Client client, Supplier<ClusterState> clusterState) {
|
||||
public PlanExecutor(Client client, Catalog catalog) {
|
||||
super(client.settings());
|
||||
|
||||
this.client = client;
|
||||
this.catalog = new EsCatalog(clusterState);
|
||||
this.catalog = catalog;
|
||||
|
||||
this.parser = new SqlParser();
|
||||
this.functionRegistry = new DefaultFunctionRegistry();
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@
|
|||
package org.elasticsearch.xpack.sql.plan.logical.command;
|
||||
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.Catalog;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsIndex;
|
||||
import org.elasticsearch.xpack.sql.expression.Attribute;
|
||||
import org.elasticsearch.xpack.sql.expression.RootFieldAttribute;
|
||||
import org.elasticsearch.xpack.sql.session.RowSetCursor;
|
||||
|
|
@ -46,10 +47,11 @@ public class ShowColumns extends Command {
|
|||
@Override
|
||||
protected RowSetCursor execute(SqlSession session) {
|
||||
Catalog catalog = session.catalog();
|
||||
Map<String, DataType> mapping = catalog.getIndex(index).mapping();
|
||||
|
||||
List<List<?>> rows = new ArrayList<>();
|
||||
fillInRows(mapping, null, rows);
|
||||
EsIndex fetched = catalog.getIndex(index);
|
||||
if (fetched != null) {
|
||||
fillInRows(fetched.mapping(), null, rows);
|
||||
}
|
||||
return Rows.of(output(), rows);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -11,21 +11,17 @@ import org.elasticsearch.client.Client;
|
|||
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNodes;
|
||||
import org.elasticsearch.cluster.service.ClusterService;
|
||||
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
|
||||
import org.elasticsearch.common.Nullable;
|
||||
import org.elasticsearch.common.settings.ClusterSettings;
|
||||
import org.elasticsearch.common.settings.IndexScopedSettings;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.common.settings.SettingsFilter;
|
||||
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
|
||||
import org.elasticsearch.env.Environment;
|
||||
import org.elasticsearch.env.NodeEnvironment;
|
||||
import org.elasticsearch.plugins.ActionPlugin;
|
||||
import org.elasticsearch.plugins.Plugin;
|
||||
import org.elasticsearch.rest.RestController;
|
||||
import org.elasticsearch.rest.RestHandler;
|
||||
import org.elasticsearch.script.ScriptService;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.watcher.ResourceWatcherService;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.Catalog;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsCatalog;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.FilteredCatalog;
|
||||
import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
||||
import org.elasticsearch.xpack.sql.plugin.cli.action.CliAction;
|
||||
import org.elasticsearch.xpack.sql.plugin.cli.action.CliHttpHandler;
|
||||
|
|
@ -42,16 +38,18 @@ import java.util.Collection;
|
|||
import java.util.List;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static java.util.Collections.singleton;
|
||||
|
||||
public class SqlPlugin extends Plugin implements ActionPlugin {
|
||||
|
||||
@Override
|
||||
public Collection<Object> createComponents(Client client, ClusterService clusterService, ThreadPool threadPool,
|
||||
ResourceWatcherService resourceWatcherService, ScriptService scriptService,
|
||||
NamedXContentRegistry xContentRegistry, Environment environment,
|
||||
NodeEnvironment nodeEnvironment, NamedWriteableRegistry namedWriteableRegistry) {
|
||||
return singleton(new PlanExecutor(client, () -> clusterService.state()));
|
||||
public class SqlPlugin implements ActionPlugin {
|
||||
/**
|
||||
* Create components used by the sql plugin.
|
||||
* @param catalogFilter if non-null it is a filter for the catalog to apply security
|
||||
*/
|
||||
public Collection<Object> createComponents(Client client, ClusterService clusterService,
|
||||
@Nullable FilteredCatalog.Filter catalogFilter) {
|
||||
EsCatalog esCatalog = new EsCatalog(() -> clusterService.state());
|
||||
Catalog catalog = catalogFilter == null ? esCatalog : new FilteredCatalog(esCatalog, catalogFilter);
|
||||
return Arrays.asList(
|
||||
esCatalog, // Added as a component so that it can get IndexNameExpressionResolver injected.
|
||||
new PlanExecutor(client, catalog));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -63,7 +61,7 @@ public class SqlPlugin extends Plugin implements ActionPlugin {
|
|||
new CliHttpHandler(settings, restController),
|
||||
new JdbcHttpHandler(settings, restController));
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
|
||||
return Arrays.asList(new ActionHandler<>(SqlAction.INSTANCE, TransportSqlAction.class),
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ import org.elasticsearch.common.inject.Inject;
|
|||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsCatalog;
|
||||
import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
||||
import org.elasticsearch.xpack.sql.plugin.cli.CliServer;
|
||||
|
||||
|
|
@ -33,9 +32,6 @@ public class TransportCliAction extends HandledTransportAction<CliRequest, CliRe
|
|||
PlanExecutor planExecutor) {
|
||||
super(settings, CliAction.NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, CliRequest::new);
|
||||
|
||||
// lazy init of the resolver
|
||||
// NOCOMMIT indexNameExpressionResolver should be available some other way; so should the localNode name :)
|
||||
((EsCatalog) planExecutor.catalog()).setIndexNameExpressionResolver(indexNameExpressionResolver);
|
||||
this.cliServer = new CliServer(planExecutor, clusterService.getClusterName().value(), () -> clusterService.localNode().getName(), Version.CURRENT, Build.CURRENT);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ import org.elasticsearch.common.inject.Inject;
|
|||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsCatalog;
|
||||
import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
||||
import org.elasticsearch.xpack.sql.plugin.jdbc.JdbcServer;
|
||||
|
||||
|
|
@ -33,9 +32,6 @@ public class TransportJdbcAction extends HandledTransportAction<JdbcRequest, Jdb
|
|||
PlanExecutor planExecutor) {
|
||||
super(settings, JdbcAction.NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, JdbcRequest::new);
|
||||
|
||||
// lazy init of the resolver
|
||||
((EsCatalog) planExecutor.catalog()).setIndexNameExpressionResolver(indexNameExpressionResolver);
|
||||
// NOCOMMIT indexNameExpressionResolver should be available some other way
|
||||
this.jdbcServer = new JdbcServer(planExecutor, clusterService.getClusterName().value(), () -> clusterService.localNode().getName(), Version.CURRENT, Build.CURRENT);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ import org.elasticsearch.common.unit.TimeValue;
|
|||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.EsCatalog;
|
||||
import org.elasticsearch.xpack.sql.execution.PlanExecutor;
|
||||
import org.elasticsearch.xpack.sql.session.RowSetCursor;
|
||||
|
||||
|
|
@ -49,9 +48,6 @@ public class TransportSqlAction extends HandledTransportAction<SqlRequest, SqlRe
|
|||
super(settings, SqlAction.NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, SqlRequest::new);
|
||||
|
||||
this.planExecutor = planExecutor;
|
||||
// lazy init of the resolver
|
||||
((EsCatalog) planExecutor.catalog()).setIndexNameExpressionResolver(indexNameExpressionResolver);
|
||||
|
||||
ephemeralId = () -> transportService.getLocalNode().getEphemeralId();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,77 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.analysis.catalog;
|
||||
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.test.ESTestCase;
|
||||
import org.elasticsearch.xpack.sql.analysis.catalog.FilteredCatalog.Filter;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static java.util.Collections.emptyList;
|
||||
import static java.util.Collections.singletonMap;
|
||||
import static java.util.stream.Collectors.toList;
|
||||
|
||||
public class FilteredCatalogTests extends ESTestCase {
|
||||
public void testGetTypeNoopCatalog() {
|
||||
Catalog orig = inMemoryCatalog("a", "b", "c");
|
||||
Catalog filtered = new FilteredCatalog(orig, new Filter() {
|
||||
@Override
|
||||
public EsIndex filterIndex(EsIndex index) {
|
||||
return index;
|
||||
}
|
||||
});
|
||||
assertSame(orig.getIndex("a"), filtered.getIndex("a"));
|
||||
assertSame(orig.getIndex("b"), filtered.getIndex("b"));
|
||||
assertSame(orig.getIndex("c"), filtered.getIndex("c"));
|
||||
assertSame(orig.getIndex("missing"), filtered.getIndex("missing"));
|
||||
}
|
||||
|
||||
public void testGetTypeFiltering() {
|
||||
Catalog orig = inMemoryCatalog("cat", "dog");
|
||||
Catalog filtered = new FilteredCatalog(orig, new Filter() {
|
||||
@Override
|
||||
public EsIndex filterIndex(EsIndex index) {
|
||||
return index.name().equals("cat") ? index : null;
|
||||
}
|
||||
});
|
||||
assertSame(orig.getIndex("cat"), filtered.getIndex("cat"));
|
||||
assertNull(filtered.getIndex("dog"));
|
||||
}
|
||||
|
||||
public void testGetTypeModifying() {
|
||||
Catalog orig = inMemoryCatalog("cat", "dog");
|
||||
Catalog filtered = new FilteredCatalog(orig, new Filter() {
|
||||
@Override
|
||||
public EsIndex filterIndex(EsIndex index) {
|
||||
if (index.name().equals("cat")) {
|
||||
return index;
|
||||
}
|
||||
return new EsIndex("rat", index.mapping(), index.aliases(), index.settings());
|
||||
}
|
||||
});
|
||||
assertSame(orig.getIndex("cat"), filtered.getIndex("cat"));
|
||||
assertEquals("rat", filtered.getIndex("dog").name());
|
||||
}
|
||||
|
||||
public void testGetTypeFilterIgnoresNull() {
|
||||
Catalog filtered = new FilteredCatalog(inMemoryCatalog(), new Filter() {
|
||||
@Override
|
||||
public EsIndex filterIndex(EsIndex index) {
|
||||
return index.name().equals("cat") ? index : null;
|
||||
}
|
||||
});
|
||||
assertNull(filtered.getIndex("missing"));
|
||||
}
|
||||
|
||||
private Catalog inMemoryCatalog(String... indexNames) {
|
||||
List<EsIndex> indices = Arrays.stream(indexNames)
|
||||
.map(i -> new EsIndex(i, singletonMap("field", null), emptyList(), Settings.EMPTY))
|
||||
.collect(toList());
|
||||
return new InMemoryCatalog(indices);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
package org.elasticsearch.xpack.sql.analysis.catalog;
|
||||
|
||||
import org.elasticsearch.xpack.sql.util.StringUtils;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.Function;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static java.util.stream.Collectors.toMap;
|
||||
|
||||
/**
|
||||
* In memory implementation of catalog designed for testing.
|
||||
*/
|
||||
class InMemoryCatalog implements Catalog {
|
||||
private final Map<String, EsIndex> indices;
|
||||
|
||||
InMemoryCatalog(List<EsIndex> indices) {
|
||||
this.indices = indices.stream().collect(toMap(EsIndex::name, Function.identity()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<EsIndex> listIndices(String pattern) {
|
||||
Pattern p = StringUtils.likeRegex(pattern);
|
||||
return indices.entrySet().stream()
|
||||
.filter(e -> p.matcher(e.getKey()).matches())
|
||||
.map(Map.Entry::getValue)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean indexIsValid(String index) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public EsIndex getIndex(String index) {
|
||||
return indices.get(index);
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue