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 f59f1e74c312bee5eebe638e227665111498d1cd Author: Li Yang <[email protected]> AuthorDate: Mon Oct 5 18:10:23 2026 +0800 KYLIN-6118 require service token on streaming internal RPC endpoints The streaming progress-report and dataflow RPC handlers (spark, stats, segment add/update/delete, and layout update) are public in the security chain and relied only on the optimistic jobExecutionId checkToken, which is not authentication. Extend the internal service-token gate to these handlers and have the streaming sender attach the deployment-scoped credential, so streaming jobs keep reporting without configuration. --- .../rest/controller/StreamingJobController.java | 22 ++++++++--- .../controller/StreamingJobControllerTest.java | 44 +++++++++++++++++++--- .../apache/kylin/streaming/rest/RestSupport.java | 5 +++ 3 files changed, 59 insertions(+), 12 deletions(-) diff --git a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/StreamingJobController.java b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/StreamingJobController.java index 6edb8a4ca6..d8ca41f135 100644 --- a/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/StreamingJobController.java +++ b/src/data-loading-server/src/main/java/org/apache/kylin/rest/controller/StreamingJobController.java @@ -30,6 +30,7 @@ import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; +import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; import org.apache.kylin.common.exception.KylinException; @@ -47,6 +48,7 @@ import org.apache.kylin.rest.response.DataResult; import org.apache.kylin.rest.response.EnvelopeResponse; import org.apache.kylin.rest.response.StreamingJobDataStatsResponse; import org.apache.kylin.rest.response.StreamingJobResponse; +import org.apache.kylin.rest.security.InternalRpcSecurity; import org.apache.kylin.rest.service.StreamingJobService; import org.apache.kylin.streaming.request.LayoutUpdateRequest; import org.apache.kylin.streaming.request.StreamingJobStatsRequest; @@ -156,7 +158,8 @@ public class StreamingJobController extends NBasicController { @PutMapping(value = "/stats") @ResponseBody public EnvelopeResponse<String> collectStreamingJobStats( - @RequestBody StreamingJobStatsRequest streamingJobStatsRequest) { + @RequestBody StreamingJobStatsRequest streamingJobStatsRequest, HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); checkProjectName(streamingJobStatsRequest.getProject()); val jobId = streamingJobStatsRequest.getJobId(); @@ -174,7 +177,8 @@ public class StreamingJobController extends NBasicController { @PutMapping(value = "/spark") @ResponseBody public EnvelopeResponse<String> updateStreamingJobInfo( - @RequestBody StreamingJobUpdateRequest streamingJobUpdateRequest) { + @RequestBody StreamingJobUpdateRequest streamingJobUpdateRequest, HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); checkProjectName(streamingJobUpdateRequest.getProject()); val meta = streamingJobService.updateStreamingJobInfo(streamingJobUpdateRequest); @@ -199,7 +203,8 @@ public class StreamingJobController extends NBasicController { */ @PostMapping(value = "/dataflow/segment") @ResponseBody - public RestResponse addSegment(@RequestBody StreamingSegmentRequest request) { + public RestResponse addSegment(@RequestBody StreamingSegmentRequest request, HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); String project = request.getProject(); String dataflowId = request.getDataflowId(); @@ -218,7 +223,9 @@ public class StreamingJobController extends NBasicController { */ @PutMapping(value = "/dataflow/segment") @ResponseBody - public RestResponse updateSegment(@RequestBody StreamingSegmentRequest request) { + public RestResponse updateSegment(@RequestBody StreamingSegmentRequest request, + HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); String project = request.getProject(); String dataflowId = request.getDataflowId(); @@ -239,7 +246,9 @@ public class StreamingJobController extends NBasicController { */ @PostMapping(value = "/dataflow/segment/deletion") @ResponseBody - public RestResponse deleteSegment(@RequestBody StreamingSegmentRequest request) { + public RestResponse deleteSegment(@RequestBody StreamingSegmentRequest request, + HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); String project = request.getProject(); String dataflowId = request.getDataflowId(); @@ -257,7 +266,8 @@ public class StreamingJobController extends NBasicController { */ @PutMapping(value = "/dataflow/layout") @ResponseBody - public RestResponse updateLayout(@RequestBody LayoutUpdateRequest request) { + public RestResponse updateLayout(@RequestBody LayoutUpdateRequest request, HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkStreamingEnabled(); String project = request.getProject(); String dataflowId = request.getDataflowId(); diff --git a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/StreamingJobControllerTest.java b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/StreamingJobControllerTest.java index b483af5b34..10e862a07d 100644 --- a/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/StreamingJobControllerTest.java +++ b/src/data-loading-server/src/test/java/org/apache/kylin/rest/controller/StreamingJobControllerTest.java @@ -27,6 +27,8 @@ import java.util.HashMap; import org.apache.commons.lang3.StringUtils; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.SystemPropertiesCache; +import org.apache.kylin.common.exception.KylinException; +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.job.execution.JobTypeEnum; @@ -56,6 +58,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; @@ -188,7 +191,8 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).collectStreamingJobStats(Mockito.any(StreamingJobStatsRequest.class)); + Mockito.verify(streamingJobController).collectStreamingJobStats(Mockito.any(StreamingJobStatsRequest.class), + Mockito.any()); } @Test @@ -204,7 +208,8 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).updateStreamingJobInfo(Mockito.any(StreamingJobUpdateRequest.class)); + Mockito.verify(streamingJobController).updateStreamingJobInfo(Mockito.any(StreamingJobUpdateRequest.class), + Mockito.any()); } @Test @@ -251,7 +256,7 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).addSegment(Mockito.any(StreamingSegmentRequest.class)); + Mockito.verify(streamingJobController).addSegment(Mockito.any(StreamingSegmentRequest.class), Mockito.any()); } @Test @@ -272,7 +277,7 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).updateSegment(Mockito.any(StreamingSegmentRequest.class)); + Mockito.verify(streamingJobController).updateSegment(Mockito.any(StreamingSegmentRequest.class), Mockito.any()); } @Test @@ -295,7 +300,7 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).deleteSegment(Mockito.any(StreamingSegmentRequest.class)); + Mockito.verify(streamingJobController).deleteSegment(Mockito.any(StreamingSegmentRequest.class), Mockito.any()); } @Test @@ -321,10 +326,37 @@ public class StreamingJobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isOk()); - Mockito.verify(streamingJobController).updateLayout(Mockito.any(LayoutUpdateRequest.class)); + Mockito.verify(streamingJobController).updateLayout(Mockito.any(LayoutUpdateRequest.class), Mockito.any()); } + @Test + public void testUpdateStreamingJobInfoAcceptsServiceToken() throws Exception { + SecurityContextHolder.clearContext(); + val request = new StreamingJobUpdateRequest(); + request.setJobType(JobTypeEnum.STREAMING_BUILD.name()); + request.setProject(PROJECT); + request.setModelId(MODEL_ID); + request.setProcessId("9921"); + Mockito.when(streamingJobService.updateStreamingJobInfo(Mockito.any())).thenReturn(new StreamingJobMeta()); + mockMvc.perform(MockMvcRequestBuilders.put("/api/streaming_jobs/spark") + .header(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, getTestConfig().getBroadcastToken()) + .contentType(MediaType.APPLICATION_JSON).content(JsonUtil.writeValueAsString(request)) + .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) + .andExpect(MockMvcResultMatchers.status().isOk()); + Mockito.verify(streamingJobController).updateStreamingJobInfo(Mockito.any(StreamingJobUpdateRequest.class), + Mockito.any()); + } + + @Test + public void testUpdateStreamingJobInfoRejectsAnonymous() { + SecurityContextHolder.clearContext(); + val request = new StreamingJobUpdateRequest(); + request.setProject(PROJECT); + Assert.assertThrows(KylinException.class, + () -> streamingJobController.updateStreamingJobInfo(request, new MockHttpServletRequest())); + } + @Test public void testGetStreamingJobDriverLogSimple() throws Exception { String job_id = "f6ca1ce7-43fc-4c42-a057-1e95dfb75d92_build"; diff --git a/src/streaming/src/main/java/org/apache/kylin/streaming/rest/RestSupport.java b/src/streaming/src/main/java/org/apache/kylin/streaming/rest/RestSupport.java index b76f906ffa..f230ac8af6 100644 --- a/src/streaming/src/main/java/org/apache/kylin/streaming/rest/RestSupport.java +++ b/src/streaming/src/main/java/org/apache/kylin/streaming/rest/RestSupport.java @@ -36,6 +36,7 @@ import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.common.response.RestResponse; import org.apache.kylin.common.util.JsonUtil; import org.apache.kylin.streaming.constants.StreamingConstants; @@ -192,5 +193,9 @@ public class RestSupport implements Closeable { private void addHeader(HttpEntityEnclosingRequestBase httpReqBase) { httpReqBase.addHeader("Accept", HTTP_VND_APACHE_KYLIN_JSON); httpReqBase.addHeader(HttpHeaders.CONTENT_TYPE, HTTP_VND_APACHE_KYLIN_JSON); + String broadcastToken = KylinConfig.getInstanceFromEnv().getBroadcastToken(); + if (StringUtils.isNotBlank(broadcastToken)) { + httpReqBase.addHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, broadcastToken); + } } }
