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 018327cdd021b12cb584563d958d1c1843282d29 Author: Li Yang <[email protected]> AuthorDate: Sat Nov 7 14:05:22 2026 +0800 KYLIN-6121 secure async query cleanup RPC --- .../apache/kylin/rest/service/RouteService.java | 2 +- .../kylin/rest/service/RouteServiceTest.java | 9 +++++++ .../rest/controller/NAsyncQueryController.java | 4 ++- .../rest/controller/NAsyncQueryControllerTest.java | 31 ++++++++++++++++++++-- 4 files changed, 42 insertions(+), 4 deletions(-) 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 728ff87e74..869e1a2480 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 @@ -113,7 +113,7 @@ public class RouteService extends BasicService { throws Exception { String uri = StringUtils.stripEnd(request.getRequestURI(), "/"); val url = "http://"; + instance + uri + "/tenant_node"; - val response = generateTaskForRemoteHost(request, url); + val response = generateTaskForRemoteHost(request, url, new byte[0], getInternalRpcHeaders()); val data = (Boolean) response.getData(); log.info("deleteAllFolder instance[{}] result : [{}]", instance, data); if (data) { diff --git a/src/common-service/src/test/java/org/apache/kylin/rest/service/RouteServiceTest.java b/src/common-service/src/test/java/org/apache/kylin/rest/service/RouteServiceTest.java index 23c4343c2c..819d4a5e13 100644 --- a/src/common-service/src/test/java/org/apache/kylin/rest/service/RouteServiceTest.java +++ b/src/common-service/src/test/java/org/apache/kylin/rest/service/RouteServiceTest.java @@ -39,6 +39,7 @@ import org.apache.kylin.common.KapConfig; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.exception.KylinException; 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.response.RestResponse; import org.apache.kylin.common.util.ExecutorServiceUtil; @@ -69,6 +70,7 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.mockito.ArgumentMatchers; import org.mockito.InjectMocks; import org.mockito.Mock; @@ -77,6 +79,7 @@ import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; +import org.springframework.http.HttpEntity; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; @@ -173,8 +176,14 @@ public class RouteServiceTest { public void deleteAllFolder() throws Exception { val result = new CountDownLatch(1); val request = new MockHttpServletRequest(); + Mockito.when(kylinConfig.getBroadcastToken()).thenReturn("service-token"); routeService.deleteAllFolder("127.0.0.1:8080", request, result); Assert.assertEquals(0, result.getCount()); + ArgumentCaptor<HttpEntity> requestEntity = ArgumentCaptor.forClass(HttpEntity.class); + Mockito.verify(restTemplate).exchange(ArgumentMatchers.anyString(), ArgumentMatchers.any(HttpMethod.class), + requestEntity.capture(), ArgumentMatchers.<Class<byte[]>> any()); + Assert.assertEquals("service-token", + requestEntity.getValue().getHeaders().getFirst(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER)); } @Test diff --git a/src/query-server/src/main/java/org/apache/kylin/rest/controller/NAsyncQueryController.java b/src/query-server/src/main/java/org/apache/kylin/rest/controller/NAsyncQueryController.java index 620069975b..c50283b708 100644 --- a/src/query-server/src/main/java/org/apache/kylin/rest/controller/NAsyncQueryController.java +++ b/src/query-server/src/main/java/org/apache/kylin/rest/controller/NAsyncQueryController.java @@ -57,6 +57,7 @@ import org.apache.kylin.rest.request.AsyncQuerySQLRequest; import org.apache.kylin.rest.response.AsyncQueryResponse; import org.apache.kylin.rest.response.EnvelopeResponse; import org.apache.kylin.rest.response.SQLResponse; +import org.apache.kylin.rest.security.InternalRpcSecurity; import org.apache.kylin.rest.service.AsyncQueryService; import org.apache.kylin.rest.service.QueryService; import org.apache.kylin.rest.util.AclEvaluate; @@ -230,7 +231,8 @@ public class NAsyncQueryController extends NBasicController { @ApiOperation(value = "delete all folder", tags = { "QE" }) @DeleteMapping(value = "/async_query/tenant_node") @ResponseBody - public EnvelopeResponse<Boolean> deleteAllFolder() throws Exception { + public EnvelopeResponse<Boolean> deleteAllFolder(HttpServletRequest request) throws Exception { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(request); if (asyncQueryService.deleteAllFolder()) { return new EnvelopeResponse<>(KylinException.CODE_SUCCESS, true, ""); } else { diff --git a/src/query-server/src/test/java/org/apache/kylin/rest/controller/NAsyncQueryControllerTest.java b/src/query-server/src/test/java/org/apache/kylin/rest/controller/NAsyncQueryControllerTest.java index bfad277739..3b95d4b977 100644 --- a/src/query-server/src/test/java/org/apache/kylin/rest/controller/NAsyncQueryControllerTest.java +++ b/src/query-server/src/test/java/org/apache/kylin/rest/controller/NAsyncQueryControllerTest.java @@ -39,6 +39,7 @@ import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.QueryContext; import org.apache.kylin.common.exception.KylinException; import org.apache.kylin.common.msg.MsgPicker; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.common.util.NLocalFileMetadataTestCase; import org.apache.kylin.query.util.AsyncQueryUtil; @@ -60,6 +61,7 @@ import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; import org.springframework.http.MediaType; +import org.springframework.mock.web.MockHttpServletRequest; import org.springframework.security.authentication.TestingAuthenticationToken; import org.springframework.security.core.Authentication; import org.springframework.security.core.context.SecurityContextHolder; @@ -550,27 +552,52 @@ public class NAsyncQueryControllerTest extends NLocalFileMetadataTestCase { @Test public void deleteAllFolder() throws Exception { Mockito.doReturn(true).when(asyncQueryService).deleteAllFolder(); + val request = new MockHttpServletRequest(); mockMvc.perform(MockMvcRequestBuilders.delete("/api/async_query/tenant_node") .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - val response = nAsyncQueryController.deleteAllFolder(); + val response = nAsyncQueryController.deleteAllFolder(request); Assert.assertTrue(response.getData()); } @Test public void deleteAllFolderFailed() throws Exception { Mockito.doReturn(false).when(asyncQueryService).deleteAllFolder(); + val request = new MockHttpServletRequest(); mockMvc.perform(MockMvcRequestBuilders.delete("/api/async_query/tenant_node") .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - val response = nAsyncQueryController.deleteAllFolder(); + val response = nAsyncQueryController.deleteAllFolder(request); Assert.assertFalse(response.getData()); } + @Test + public void deleteAllFolderRejectsLowPrivilegeUser() throws Exception { + Authentication otherUser = new TestingAuthenticationToken("OTHER", "OTHER", Constant.IDENTITY_USER); + SecurityContextHolder.getContext().setAuthentication(otherUser); + + Assert.assertThrows(KylinException.class, + () -> nAsyncQueryController.deleteAllFolder(new MockHttpServletRequest())); + Mockito.verify(asyncQueryService, Mockito.never()).deleteAllFolder(); + } + + @Test + public void deleteAllFolderAcceptsServiceToken() throws Exception { + Authentication otherUser = new TestingAuthenticationToken("OTHER", "OTHER", Constant.IDENTITY_USER); + SecurityContextHolder.getContext().setAuthentication(otherUser); + Mockito.doReturn(true).when(asyncQueryService).deleteAllFolder(); + val request = new MockHttpServletRequest(); + request.addHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, getTestConfig().getBroadcastToken()); + + val response = nAsyncQueryController.deleteAllFolder(request); + + Assert.assertTrue(response.getData()); + } + @Test public void testBatchDeleteOldResultWhenTimeFormatError() throws Exception { Mockito.doThrow(new ParseException("", 0)).when(asyncQueryService).batchDelete(Mockito.any(), Mockito.any(),
