This is an automated email from the ASF dual-hosted git repository. liyang-kylin pushed a commit to branch kylin5 in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 82ba881c4da629c93b6978167dcf4065e05f3326 Author: Li Yang <[email protected]> AuthorDate: Fri Oct 2 14:24:02 2026 +0800 KYLIN-6108 improve gluten cache access control --- .../kylin/rest/controller/BroadcastController.java | 22 +------- .../rest/security/BroadcastSecurityContext.java | 9 +-- ...curityContext.java => InternalRpcSecurity.java} | 64 +++++++++------------- .../apache/kylin/rest/service/BasicService.java | 13 +++++ .../apache/kylin/rest/service/RouteService.java | 10 +++- .../kylin/tool/restclient/RestClientTest.java | 5 ++ .../kylin/rest/controller/JobController.java | 12 +++- .../kylin/rest/controller/JobControllerTest.java | 7 ++- .../apache/kylin/utils/GlutenCacheUtilsTest.java | 11 ++++ .../rest/controller/GlutenCacheController.java | 21 ++++++- .../rest/controller/GlutenCacheControllerTest.java | 19 ++++++- .../kylin/rest/service/GlutenCacheService.java | 2 + .../kylin/rest/service/GlutenCacheServiceTest.java | 12 ++-- .../kylin/engine/spark/job/LoadCacheStep.java | 5 ++ .../org/apache/kylin/utils/GlutenCacheUtils.java | 30 ++++++++++ 15 files changed, 162 insertions(+), 80 deletions(-) diff --git a/src/common-server/src/main/java/org/apache/kylin/rest/controller/BroadcastController.java b/src/common-server/src/main/java/org/apache/kylin/rest/controller/BroadcastController.java index 43dc8a9a14..a6f608ff84 100644 --- a/src/common-server/src/main/java/org/apache/kylin/rest/controller/BroadcastController.java +++ b/src/common-server/src/main/java/org/apache/kylin/rest/controller/BroadcastController.java @@ -20,21 +20,17 @@ package org.apache.kylin.rest.controller; import static org.apache.kylin.common.constant.HttpConstant.HTTP_VND_APACHE_KYLIN_JSON; import static org.apache.kylin.common.constant.HttpConstant.HTTP_VND_APACHE_KYLIN_V4_PUBLIC_JSON; -import static org.apache.kylin.common.exception.ServerErrorCode.PERMISSION_DENIED; import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; import javax.servlet.http.HttpServletRequest; -import org.apache.commons.lang3.StringUtils; -import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.exception.KylinException; import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.rest.config.initialize.BroadcastListener; import org.apache.kylin.rest.response.EnvelopeResponse; import org.apache.kylin.rest.security.BroadcastSecurityContext; +import org.apache.kylin.rest.security.InternalRpcSecurity; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.PostMapping; @@ -64,21 +60,7 @@ public class BroadcastController extends NBasicController { } private void authorizeBroadcast(HttpServletRequest request) { - if (isAdmin()) { - return; - } - - String configuredToken = KylinConfig.getInstanceFromEnv().getBroadcastToken(); - String providedToken = request == null ? null - : request.getHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER); - if (StringUtils.isNotBlank(configuredToken) && providedToken != null - && MessageDigest.isEqual(configuredToken.getBytes(StandardCharsets.UTF_8), - providedToken.getBytes(StandardCharsets.UTF_8))) { - return; - } - - throw new KylinException(PERMISSION_DENIED, - "Broadcast endpoint requires an authenticated global admin or a valid broadcast token."); + InternalRpcSecurity.requireGlobalAdminOrServiceToken(request); } @PutMapping(value = "/capacity/refresh_all") diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java b/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java index 57f3563ab3..fbcafb1dbf 100644 --- a/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java +++ b/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java @@ -22,9 +22,6 @@ import static org.apache.kylin.common.exception.ServerErrorCode.PERMISSION_DENIE import java.io.IOException; import org.apache.kylin.common.exception.KylinException; -import org.apache.kylin.rest.constant.Constant; -import org.springframework.security.core.Authentication; -import org.springframework.security.core.context.SecurityContextHolder; import lombok.extern.slf4j.Slf4j; @@ -59,11 +56,7 @@ public final class BroadcastSecurityContext { return; } - Authentication authentication = SecurityContextHolder.getContext().getAuthentication(); - boolean globalAdmin = authentication != null && authentication.isAuthenticated() - && authentication.getAuthorities().stream() - .anyMatch(authority -> Constant.ROLE_ADMIN.equals(authority.getAuthority())); - if (!globalAdmin) { + if (!InternalRpcSecurity.isGlobalAdmin()) { log.warn("Rejected an untrusted broadcast-originated ACL mutation"); throw new KylinException(PERMISSION_DENIED, "Broadcast ACL mutation requires a trusted internal caller."); } diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java b/src/common-service/src/main/java/org/apache/kylin/rest/security/InternalRpcSecurity.java similarity index 52% copy from src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java copy to src/common-service/src/main/java/org/apache/kylin/rest/security/InternalRpcSecurity.java index 57f3563ab3..84bcab9878 100644 --- a/src/common-service/src/main/java/org/apache/kylin/rest/security/BroadcastSecurityContext.java +++ b/src/common-service/src/main/java/org/apache/kylin/rest/security/InternalRpcSecurity.java @@ -19,58 +19,44 @@ package org.apache.kylin.rest.security; import static org.apache.kylin.common.exception.ServerErrorCode.PERMISSION_DENIED; -import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import javax.servlet.http.HttpServletRequest; + +import org.apache.commons.lang3.StringUtils; +import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.exception.KylinException; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.rest.constant.Constant; import org.springframework.security.core.Authentication; import org.springframework.security.core.context.SecurityContextHolder; -import lombok.extern.slf4j.Slf4j; - -@Slf4j -public final class BroadcastSecurityContext { - - private static final ThreadLocal<Boolean> TRUSTED_BROADCAST = new ThreadLocal<>(); - - private BroadcastSecurityContext() { - } - - public static boolean isTrustedBroadcast() { - return Boolean.TRUE.equals(TRUSTED_BROADCAST.get()); - } +public final class InternalRpcSecurity { - public static void runAsTrusted(CheckedRunnable action) throws IOException { - boolean previous = isTrustedBroadcast(); - TRUSTED_BROADCAST.set(Boolean.TRUE); - try { - action.run(); - } finally { - if (previous) { - TRUSTED_BROADCAST.set(Boolean.TRUE); - } else { - TRUSTED_BROADCAST.remove(); - } - } + private InternalRpcSecurity() { } - public static void requireTrustedBroadcastOrGlobalAdmin() { - if (isTrustedBroadcast()) { - return; - } - + public static boolean isGlobalAdmin() { Authentication authentication = SecurityContextHolder.getContext().getAuthentication(); - boolean globalAdmin = authentication != null && authentication.isAuthenticated() + return authentication != null && authentication.isAuthenticated() && authentication.getAuthorities().stream() .anyMatch(authority -> Constant.ROLE_ADMIN.equals(authority.getAuthority())); - if (!globalAdmin) { - log.warn("Rejected an untrusted broadcast-originated ACL mutation"); - throw new KylinException(PERMISSION_DENIED, "Broadcast ACL mutation requires a trusted internal caller."); - } } - @FunctionalInterface - public interface CheckedRunnable { - void run() throws IOException; + public static boolean hasValidServiceToken(HttpServletRequest request) { + String configuredToken = KylinConfig.getInstanceFromEnv().getBroadcastToken(); + String providedToken = request == null ? null + : request.getHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER); + return StringUtils.isNotBlank(configuredToken) && providedToken != null + && MessageDigest.isEqual(configuredToken.getBytes(StandardCharsets.UTF_8), + providedToken.getBytes(StandardCharsets.UTF_8)); + } + + public static void requireGlobalAdminOrServiceToken(HttpServletRequest request) { + if (!isGlobalAdmin() && !hasValidServiceToken(request)) { + throw new KylinException(PERMISSION_DENIED, + "Internal RPC requires an authenticated global admin or a valid service token."); + } } } diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/service/BasicService.java b/src/common-service/src/main/java/org/apache/kylin/rest/service/BasicService.java index ec5ac3d401..e7302865ec 100644 --- a/src/common-service/src/main/java/org/apache/kylin/rest/service/BasicService.java +++ b/src/common-service/src/main/java/org/apache/kylin/rest/service/BasicService.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Set; import javax.servlet.http.HttpServletRequest; @@ -126,12 +127,24 @@ public abstract class BasicService { return JsonUtil.readValue(response.getBody(), EnvelopeResponse.class); } + public <T> EnvelopeResponse<T> generateTaskForRemoteHost(final HttpServletRequest request, String url, byte[] body, + Map<String, String> additionalHeaders) throws Exception { + val response = getHttpResponse(request, url, body, additionalHeaders); + return JsonUtil.readValue(response.getBody(), EnvelopeResponse.class); + } + private ResponseEntity<byte[]> getHttpResponse(final HttpServletRequest request, String url, byte[] body) { + return getHttpResponse(request, url, body, Collections.emptyMap()); + } + + private ResponseEntity<byte[]> getHttpResponse(final HttpServletRequest request, String url, byte[] body, + Map<String, String> additionalHeaders) { HttpHeaders headers = new HttpHeaders(); Collections.list(request.getHeaderNames()) .forEach(k -> headers.put(k, Collections.list(request.getHeaders(k)))); //remove gzip headers.remove(ACCEPT_ENCODING); + additionalHeaders.forEach(headers::set); return restTemplate.exchange(url, HttpMethod.valueOf(request.getMethod()), new HttpEntity<>(body, headers), byte[].class); } diff --git a/src/common-service/src/main/java/org/apache/kylin/rest/service/RouteService.java b/src/common-service/src/main/java/org/apache/kylin/rest/service/RouteService.java index 7bf6a1cda2..728ff87e74 100644 --- a/src/common-service/src/main/java/org/apache/kylin/rest/service/RouteService.java +++ b/src/common-service/src/main/java/org/apache/kylin/rest/service/RouteService.java @@ -20,6 +20,7 @@ package org.apache.kylin.rest.service; import static org.apache.kylin.common.exception.ServerErrorCode.PERMISSION_DENIED; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -42,6 +43,7 @@ import org.apache.kylin.common.constant.LogConstant; import org.apache.kylin.common.exception.KylinException; import org.apache.kylin.common.logging.SetLogCategory; import org.apache.kylin.common.msg.MsgPicker; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.common.persistence.transaction.UnitOfWork; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.common.util.NamedThreadFactory; @@ -270,7 +272,7 @@ public class RouteService extends BasicService { CountDownLatch result) throws Exception { try (SetLogCategory ignore = new SetLogCategory(LogConstant.BUILD_CATEGORY)) { val fullUrl = "http://"; + instance + url; - val response = generateTaskForRemoteHost(request, fullUrl, requestEntity); + val response = generateTaskForRemoteHost(request, fullUrl, requestEntity, getInternalRpcHeaders()); log.info("cacheGluten instance is [{}], result is [{}]", instance, response); if (StringUtils.equals(url, CACHE_GLUTEN_API)) { val data = JsonUtil.convert(response.getData(), GlutenCacheResponse.class); @@ -284,6 +286,12 @@ public class RouteService extends BasicService { } } + private Map<String, String> getInternalRpcHeaders() { + String token = KylinConfig.getInstanceFromEnv().getBroadcastToken(); + return StringUtils.isBlank(token) ? Collections.emptyMap() + : Collections.singletonMap(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, token); + } + public boolean checkGlutenCacheLimits(Map<Future<?>, Long> tasks) { var limitFlag = false; for (Future<?> doneTask : tasks.keySet()) { diff --git a/src/core-common/src/test/java/org/apache/kylin/tool/restclient/RestClientTest.java b/src/core-common/src/test/java/org/apache/kylin/tool/restclient/RestClientTest.java index 28f508b357..95f3510dbf 100644 --- a/src/core-common/src/test/java/org/apache/kylin/tool/restclient/RestClientTest.java +++ b/src/core-common/src/test/java/org/apache/kylin/tool/restclient/RestClientTest.java @@ -46,6 +46,11 @@ public class RestClientTest extends NLocalFileMetadataTestCase { cleanupTestMetadata(); } + @Test + public void testNoDefaultBroadcastCredential() { + Assert.assertEquals("", getTestConfig().getBroadcastToken()); + } + @Test public void testNotifyAddsBroadcastToken() throws Exception { getTestConfig().setProperty("kylin.server.broadcast-token", "secret"); diff --git a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java index dfba66a6d3..26bf843c1d 100644 --- a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java +++ b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/JobController.java @@ -59,10 +59,13 @@ import org.apache.kylin.rest.response.EventResponse; import org.apache.kylin.rest.response.ExecutableResponse; import org.apache.kylin.rest.response.ExecutableStepResponse; import org.apache.kylin.rest.response.JobStatisticsResponse; +import org.apache.kylin.rest.security.InternalRpcSecurity; import org.apache.kylin.rest.service.JobInfoService; import org.apache.kylin.rest.service.JobService; import org.apache.kylin.rest.service.RouteService; +import org.apache.kylin.rest.util.AclEvaluate; import org.apache.kylin.rest.util.SparkUIUtil; +import org.apache.kylin.utils.GlutenCacheUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -101,6 +104,9 @@ public class JobController extends BaseController { @Autowired private RouteService routeService; + @Autowired + private AclEvaluate aclEvaluate; + @Override protected Logger getLogger() { return logger; @@ -488,8 +494,12 @@ public class JobController extends BaseController { public EnvelopeResponse<Boolean> routeGlutenCache(@RequestBody LoadGlutenCacheRequest request, HttpServletRequest servletRequest) throws Exception { logger.info("routeGlutenCache request is [{}]", request); - checkProjectName(request.getProject()); + val projectName = checkProjectName(request.getProject()); + if (!InternalRpcSecurity.isGlobalAdmin() && !InternalRpcSecurity.hasValidServiceToken(servletRequest)) { + aclEvaluate.checkProjectAdminPermission(projectName); + } checkCollectionRequiredArg("cache_commands", request.getCacheCommands()); + GlutenCacheUtils.validateCacheCommands(request.getCacheCommands()); val result = routeService.routeGlutenCache(request.getCacheCommands(), servletRequest); return new EnvelopeResponse<>(KylinException.CODE_SUCCESS, result, ""); } diff --git a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java index 9eb47f0214..0285ba5a4e 100644 --- a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java +++ b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/JobControllerTest.java @@ -58,6 +58,7 @@ import org.apache.kylin.rest.response.ExecutableStepResponse; import org.apache.kylin.rest.service.JobInfoService; import org.apache.kylin.rest.service.JobService; import org.apache.kylin.rest.service.RouteService; +import org.apache.kylin.rest.util.AclEvaluate; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -97,6 +98,9 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { @Mock private JobInfoService jobInfoService; + @Mock + private AclEvaluate aclEvaluate; + @InjectMocks private final JobController jobController = Mockito.spy(new JobController()); @@ -642,7 +646,8 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { @Test public void routeGlutenCache() throws Exception { val servletRequest = new MockHttpServletRequest(); - val request = new LoadGlutenCacheRequest("default", Lists.newArrayList("test command")); + val request = new LoadGlutenCacheRequest("default", + Lists.newArrayList("CACHE DATA SELECT * FROM 'test-path'")); Mockito.when(routeService.routeGlutenCache(request.getCacheCommands(), servletRequest)).thenReturn(true); val mvcResult = mockMvc.perform(MockMvcRequestBuilders.post("/api/jobs/gluten_cache") // .contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request))) diff --git a/src/data-loading-service/src/test/java/org/apache/kylin/utils/GlutenCacheUtilsTest.java b/src/data-loading-service/src/test/java/org/apache/kylin/utils/GlutenCacheUtilsTest.java index b63514a716..5390ab03b3 100644 --- a/src/data-loading-service/src/test/java/org/apache/kylin/utils/GlutenCacheUtilsTest.java +++ b/src/data-loading-service/src/test/java/org/apache/kylin/utils/GlutenCacheUtilsTest.java @@ -137,6 +137,17 @@ class GlutenCacheUtilsTest extends AbstractTestCase { generateTableCommandError(config); } + @Test + void validateCacheCommands() { + Assertions.assertDoesNotThrow(() -> GlutenCacheUtils.validateCacheCommands(Lists.newArrayList( + "CACHE DATA SELECT * FROM 'test-path'", + "CACHE FILES ASYNC SELECT * FROM 'test-path' CACHEPROPERTIES (recursive=true)"))); + Assertions.assertThrows(KylinRuntimeException.class, + () -> GlutenCacheUtils.validateCacheCommands(Lists.newArrayList("DROP TABLE test"))); + Assertions.assertThrows(KylinRuntimeException.class, () -> GlutenCacheUtils + .validateCacheCommands(Lists.newArrayList("CACHE DATA SELECT * FROM 'test'; DROP TABLE test"))); + } + private static void generateTableCommand(KylinConfig config, String location, InternalTableDesc internalTable, String datePartitionFormat) { var start = ""; diff --git a/src/query-server/src/main/java/org/apache/kylin/rest/controller/GlutenCacheController.java b/src/query-server/src/main/java/org/apache/kylin/rest/controller/GlutenCacheController.java index 90e23213f0..a605bac0b5 100644 --- a/src/query-server/src/main/java/org/apache/kylin/rest/controller/GlutenCacheController.java +++ b/src/query-server/src/main/java/org/apache/kylin/rest/controller/GlutenCacheController.java @@ -30,7 +30,9 @@ import org.apache.kylin.rest.request.IndexGlutenCacheRequest; import org.apache.kylin.rest.request.InternalTableGlutenCacheRequest; import org.apache.kylin.rest.response.EnvelopeResponse; import org.apache.kylin.rest.response.GlutenCacheResponse; +import org.apache.kylin.rest.security.InternalRpcSecurity; import org.apache.kylin.rest.service.GlutenCacheService; +import org.apache.kylin.rest.util.AclEvaluate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -50,13 +52,18 @@ public class GlutenCacheController extends NBasicController { @Autowired private GlutenCacheService glutenCacheService; + @Autowired + private AclEvaluate aclEvaluate; + /** * RPC Call: for build job update gluten cache */ @PostMapping(value = "/cache/gluten_cache") @ApiOperation(value = "gluten cache", tags = { "Gluten" }) @ResponseBody - public EnvelopeResponse<GlutenCacheResponse> glutenCache(@RequestBody List<String> cacheCommands) { + public EnvelopeResponse<GlutenCacheResponse> glutenCache(@RequestBody List<String> cacheCommands, + HttpServletRequest request) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(request); checkCollectionRequiredArg("cacheCommands", cacheCommands); val result = glutenCacheService.glutenCache(cacheCommands); return new EnvelopeResponse<>(KylinException.CODE_SUCCESS, result, ""); @@ -68,7 +75,9 @@ public class GlutenCacheController extends NBasicController { @PostMapping(value = "/cache/gluten_cache_async") @ApiOperation(value = "gluten cache", tags = { "Gluten" }) @ResponseBody - public EnvelopeResponse<String> glutenCacheAsync(@RequestBody List<String> cacheCommands) { + public EnvelopeResponse<String> glutenCacheAsync(@RequestBody List<String> cacheCommands, + HttpServletRequest request) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(request); checkCollectionRequiredArg("cacheCommands", cacheCommands); glutenCacheService.glutenCacheAsync(cacheCommands); return new EnvelopeResponse<>(KylinException.CODE_SUCCESS, "", ""); @@ -81,6 +90,7 @@ public class GlutenCacheController extends NBasicController { HttpServletRequest servletRequest) throws Exception { log.info("Internal table gluten cache request is [{}]", request); val projectName = checkProjectName(request.getProject()); + checkProjectAdminPermission(projectName, servletRequest); request.setProject(projectName); checkRequiredArg("database", request.getDatabase()); checkRequiredArg("table", request.getTable()); @@ -95,9 +105,16 @@ public class GlutenCacheController extends NBasicController { HttpServletRequest servletRequest) throws Exception { log.info("Index gluten cache request is [{}]", request); val projectName = checkProjectName(request.getProject()); + checkProjectAdminPermission(projectName, servletRequest); request.setProject(projectName); checkRequiredArg("model", request.getModel()); glutenCacheService.indexGlutenCache(request, servletRequest); return new EnvelopeResponse<>(KylinException.CODE_SUCCESS, "", ""); } + + private void checkProjectAdminPermission(String projectName, HttpServletRequest request) { + if (!InternalRpcSecurity.isGlobalAdmin() && !InternalRpcSecurity.hasValidServiceToken(request)) { + aclEvaluate.checkProjectAdminPermission(projectName); + } + } } diff --git a/src/query-server/src/test/java/org/apache/kylin/rest/controller/GlutenCacheControllerTest.java b/src/query-server/src/test/java/org/apache/kylin/rest/controller/GlutenCacheControllerTest.java index 8773618b08..79e432f93f 100644 --- a/src/query-server/src/test/java/org/apache/kylin/rest/controller/GlutenCacheControllerTest.java +++ b/src/query-server/src/test/java/org/apache/kylin/rest/controller/GlutenCacheControllerTest.java @@ -20,6 +20,7 @@ package org.apache.kylin.rest.controller; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.common.util.NLocalFileMetadataTestCase; +import org.apache.kylin.common.exception.KylinException; import org.apache.kylin.guava30.shaded.common.collect.Lists; import org.apache.kylin.job.util.JobContextUtil; import org.apache.kylin.rest.constant.Constant; @@ -27,7 +28,9 @@ import org.apache.kylin.rest.request.IndexGlutenCacheRequest; import org.apache.kylin.rest.request.InternalTableGlutenCacheRequest; import org.apache.kylin.rest.response.GlutenCacheResponse; import org.apache.kylin.rest.service.GlutenCacheService; +import org.apache.kylin.rest.util.AclEvaluate; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.InjectMocks; @@ -55,6 +58,9 @@ public class GlutenCacheControllerTest extends NLocalFileMetadataTestCase { @Mock private GlutenCacheService glutenCacheService; + @Mock + private AclEvaluate aclEvaluate; + @InjectMocks private GlutenCacheController glutenCacheController = Mockito.spy(new GlutenCacheController()); @@ -87,7 +93,7 @@ public class GlutenCacheControllerTest extends NLocalFileMetadataTestCase { .contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request))) .andExpect(MockMvcResultMatchers.status().isOk()).andReturn(); - Mockito.verify(glutenCacheController).glutenCache(request); + Mockito.verify(glutenCacheController).glutenCache(Mockito.eq(request), Mockito.any()); } @Test @@ -99,7 +105,7 @@ public class GlutenCacheControllerTest extends NLocalFileMetadataTestCase { .contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request))) .andExpect(MockMvcResultMatchers.status().isOk()).andReturn(); - Mockito.verify(glutenCacheController).glutenCacheAsync(request); + Mockito.verify(glutenCacheController).glutenCacheAsync(Mockito.eq(request), Mockito.any()); } @Test @@ -130,4 +136,13 @@ public class GlutenCacheControllerTest extends NLocalFileMetadataTestCase { Mockito.verify(glutenCacheController).indexGlutenCache(request, mvcResult.getRequest()); } + + @Test + public void anonymousCacheRequestRejected() { + SecurityContextHolder.clearContext(); + + Assert.assertThrows(KylinException.class, + () -> glutenCacheController.glutenCache(Lists.newArrayList("DROP TABLE test"), + new MockHttpServletRequest())); + } } diff --git a/src/query-service/src/main/java/org/apache/kylin/rest/service/GlutenCacheService.java b/src/query-service/src/main/java/org/apache/kylin/rest/service/GlutenCacheService.java index d0f95ac024..3fb853003a 100644 --- a/src/query-service/src/main/java/org/apache/kylin/rest/service/GlutenCacheService.java +++ b/src/query-service/src/main/java/org/apache/kylin/rest/service/GlutenCacheService.java @@ -68,6 +68,7 @@ public class GlutenCacheService extends BasicService { public GlutenCacheResponse glutenCache(List<String> cacheCommands) { val executeResults = Lists.<GlutenCacheExecuteResult> newArrayList(); try (SetLogCategory ignore = new SetLogCategory(LogConstant.BUILD_CATEGORY)) { + GlutenCacheUtils.validateCacheCommands(cacheCommands); val sparkSession = SparderEnv.getSparkSession(); val successCount = new AtomicInteger(0); for (String cacheCommand : cacheCommands) { @@ -172,6 +173,7 @@ public class GlutenCacheService extends BasicService { public void glutenCacheAsync(List<String> cacheCommands) { try (GlutenCacheRequestLimits ignored = new GlutenCacheRequestLimits(); SetLogCategory ignore = new SetLogCategory(LogConstant.BUILD_CATEGORY)) { + GlutenCacheUtils.validateCacheCommands(cacheCommands); val sparkSession = SparderEnv.getSparkSession(); for (String cacheCommand : cacheCommands) { glutenCacheExecute(cacheCommand, sparkSession); diff --git a/src/query-service/src/test/java/org/apache/kylin/rest/service/GlutenCacheServiceTest.java b/src/query-service/src/test/java/org/apache/kylin/rest/service/GlutenCacheServiceTest.java index 671d9363ca..12cf99447d 100644 --- a/src/query-service/src/test/java/org/apache/kylin/rest/service/GlutenCacheServiceTest.java +++ b/src/query-service/src/test/java/org/apache/kylin/rest/service/GlutenCacheServiceTest.java @@ -97,12 +97,12 @@ public class GlutenCacheServiceTest extends NLocalFileMetadataTestCase { @Mock private Appender appender = Mockito.mock(Appender.class); - private static final String COMMAND_0 = "test command 0"; - private static final String COMMAND_1 = "test command 1"; - private static final String COMMAND_2 = "test command 2"; - private static final String COMMAND_3 = "test command 3"; - private static final String COMMAND_4 = "test command 4"; - private static final String COMMAND_5 = "test command 5"; + private static final String COMMAND_0 = "CACHE DATA SELECT * FROM 'test-command-0'"; + private static final String COMMAND_1 = "CACHE DATA SELECT * FROM 'test-command-1'"; + private static final String COMMAND_2 = "CACHE DATA SELECT * FROM 'test-command-2'"; + private static final String COMMAND_3 = "CACHE DATA SELECT * FROM 'test-command-3'"; + private static final String COMMAND_4 = "CACHE DATA SELECT * FROM 'test-command-4'"; + private static final String COMMAND_5 = "CACHE DATA SELECT * FROM 'test-command-5'"; private static final String PROJECT = "default"; private static final String DATABASE = "databases"; private static final String TABLE = "table"; diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/LoadCacheStep.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/LoadCacheStep.java index 79a2dca4aa..fb3ed72d4c 100644 --- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/LoadCacheStep.java +++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/LoadCacheStep.java @@ -37,6 +37,7 @@ import org.apache.http.util.EntityUtils; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.exception.KylinException; import org.apache.kylin.common.exception.KylinRuntimeException; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.common.util.AddressUtil; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.guava30.shaded.common.collect.Maps; @@ -79,6 +80,10 @@ public abstract class LoadCacheStep extends AbstractExecutable { val httpPost = new HttpPost(requestApi); httpPost.addHeader(HttpHeaders.CONTENT_TYPE, HTTP_VND_APACHE_KYLIN_JSON); + String serviceToken = config.getBroadcastToken(); + if (StringUtils.isNotBlank(serviceToken)) { + httpPost.addHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, serviceToken); + } httpPost.setEntity(new ByteArrayEntity(requestEntity, ContentType.APPLICATION_JSON)); val httpResponse = httpClient.execute(httpPost); diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/utils/GlutenCacheUtils.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/utils/GlutenCacheUtils.java index 7f4b2106dd..9b1513fb4e 100644 --- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/utils/GlutenCacheUtils.java +++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/utils/GlutenCacheUtils.java @@ -55,6 +55,36 @@ import lombok.experimental.UtilityClass; @UtilityClass public class GlutenCacheUtils { + private static final String CACHE_DATA_PREFIX = "CACHE DATA "; + private static final String CACHE_FILES_PREFIX = "CACHE FILES "; + + public static void validateCacheCommands(Collection<String> cacheCommands) { + if (CollectionUtils.isEmpty(cacheCommands)) { + return; + } + for (String cacheCommand : cacheCommands) { + if (StringUtils.isBlank(cacheCommand)) { + throw new KylinRuntimeException("Invalid gluten cache command: command is empty"); + } + String command = cacheCommand.trim(); + if (command.indexOf(';') >= 0 || command.indexOf('\n') >= 0 || command.indexOf('\r') >= 0) { + throw new KylinRuntimeException("Invalid gluten cache command: statement separator is not allowed"); + } + if (command.startsWith(CACHE_DATA_PREFIX)) { + if (!command.contains(" SELECT ") || !command.contains(" FROM '")) { + throw new KylinRuntimeException("Invalid gluten cache data command"); + } + } else if (command.startsWith(CACHE_FILES_PREFIX)) { + if (!command.contains(" SELECT * FROM '") + || !command.endsWith(" CACHEPROPERTIES (recursive=true)")) { + throw new KylinRuntimeException("Invalid gluten cache files command"); + } + } else { + throw new KylinRuntimeException("Invalid gluten cache command"); + } + } + } + public static String generateCacheTableCommand(KylinConfig config, String project, String table, String start, List<String> columns, boolean isAsync) { val internalTableManager = InternalTableManager.getInstance(config, project);
