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 43e1890d5723ba779b7d4cbb6d97a5d723e27e5e Author: Li Yang <[email protected]> AuthorDate: Mon Oct 5 17:57:09 2026 +0800 KYLIN-6118 require service token on internal job-report RPC endpoints The job progress-report RPC handlers (error, stage status, spark, and wait/run time) are exposed on the public API port and were listed as public in the security chain without any authorization check. Only an optimistic job-status check protected them, so a caller that knows a running job identifier and start time could forge job progress metadata. Require a global admin or a valid internal service credential on all four handlers, and have the Spark driver progress reporter send the deployment-scoped credential, matching the broadcast and cache RPC fixes. --- .../kylin/rest/controller/JobController.java | 13 ++++-- .../kylin/rest/controller/JobControllerTest.java | 54 +++++++++++++++++++--- .../engine/spark/job/RestfulJobProgressReport.java | 6 +++ 3 files changed, 63 insertions(+), 10 deletions(-) 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 26bf843c1d..3234d5a0be 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 @@ -273,7 +273,9 @@ public class JobController extends BaseController { @ApiOperation(value = "updateJobError", tags = { "DW" }, notes = "Update Body: job error") @PutMapping(value = "error") @ResponseBody - public EnvelopeResponse<String> updateJobError(@RequestBody JobErrorRequest request) { + public EnvelopeResponse<String> updateJobError(@RequestBody JobErrorRequest request, + HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); if (StringUtils.isBlank(request.getProject()) && StringUtils.isBlank(request.getJobId())) { throw new KylinException(JOB_ID_EMPTY, "At least one job should be selected to update stage status"); } @@ -294,7 +296,9 @@ public class JobController extends BaseController { @ApiOperation(value = "updateStageStatus", tags = { "DW" }, notes = "Update Body: jobIds(stage ids)") @PutMapping(value = "/stage/status") @ResponseBody - public EnvelopeResponse<String> updateStageStatus(@RequestBody StageRequest stageRequest) { + public EnvelopeResponse<String> updateStageStatus(@RequestBody StageRequest stageRequest, + HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); if (StringUtils.isBlank(stageRequest.getProject()) && StringUtils.isBlank(stageRequest.getTaskId())) { throw new KylinException(JOB_ID_EMPTY, "At least one job should be selected to update stage status"); } @@ -318,6 +322,7 @@ public class JobController extends BaseController { @ResponseBody public EnvelopeResponse<String> updateSparkJobInfo(@RequestBody SparkJobUpdateRequest sparkJobUpdateRequest, HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkProjectName(sparkJobUpdateRequest.getProject()); if (StringUtils.isNotBlank(sparkJobUpdateRequest.getYarnAppUrl())) { SparkUIUtil.validateJobTrackingUrl(sparkJobUpdateRequest.getYarnAppUrl(), @@ -337,7 +342,9 @@ public class JobController extends BaseController { @PutMapping(value = "/wait_and_run_time") @ApiOperation(value = "updateWaitTime", tags = { "DW" }) @ResponseBody - public EnvelopeResponse<String> updateSparkJobTime(@RequestBody SparkJobTimeRequest sparkJobTimeRequest) { + public EnvelopeResponse<String> updateSparkJobTime(@RequestBody SparkJobTimeRequest sparkJobTimeRequest, + HttpServletRequest servletRequest) { + InternalRpcSecurity.requireGlobalAdminOrServiceToken(servletRequest); checkProjectName(sparkJobTimeRequest.getProject()); return updateJobInfoWithCheck(sparkJobTimeRequest.getProject(), sparkJobTimeRequest.getJobId(), sparkJobTimeRequest.getJobLastRunningStartTime(), 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 0285ba5a4e..ead614d1e9 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 @@ -28,6 +28,7 @@ import java.util.stream.Collectors; 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.common.util.JsonUtil; import org.apache.kylin.common.util.NLocalFileMetadataTestCase; import org.apache.kylin.common.util.Pair; @@ -401,7 +402,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { Map<String, String> response = JsonUtil.readValueAsMap(result.getResponse().getContentAsString()); Assert.assertEquals(response.get("code"), KylinException.CODE_SUCCESS); - Mockito.verify(jobController).updateJobError(request); + Mockito.verify(jobController).updateJobError(Mockito.eq(request), Mockito.any()); } @Test @@ -422,7 +423,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { Map<String, String> response = JsonUtil.readValueAsMap(result.getResponse().getContentAsString()); Assert.assertEquals(response.get("code"), KylinException.CODE_SUCCESS); - Mockito.verify(jobController).updateStageStatus(request); + Mockito.verify(jobController).updateStageStatus(Mockito.eq(request), Mockito.any()); request = new StageRequest(); request.setProject(""); @@ -437,7 +438,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) .andExpect(MockMvcResultMatchers.status().isInternalServerError()).andReturn(); - Mockito.verify(jobController).updateStageStatus(request); + Mockito.verify(jobController).updateStageStatus(Mockito.eq(request), Mockito.any()); } @Test @@ -459,7 +460,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { Assert.assertEquals(response.get("code"), KylinException.CODE_UNDEFINED); Assert.assertEquals(response.get("msg"), "Job has stopped."); - Mockito.verify(jobController).updateStageStatus(request); + Mockito.verify(jobController).updateStageStatus(Mockito.eq(request), Mockito.any()); } @Test @@ -481,7 +482,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { Assert.assertEquals(response.get("code"), KylinException.CODE_UNDEFINED); Assert.assertEquals(response.get("msg"), "Job has stopped."); - Mockito.verify(jobController).updateStageStatus(request); + Mockito.verify(jobController).updateStageStatus(Mockito.eq(request), Mockito.any()); } @Test @@ -528,7 +529,7 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { } }); - Mockito.verify(jobController, Mockito.times(repeatTime)).updateStageStatus(request); + Mockito.verify(jobController, Mockito.times(repeatTime)).updateStageStatus(Mockito.eq(request), Mockito.any()); Assert.assertEquals(0, failedCount.get()); } @@ -552,7 +553,46 @@ public class JobControllerTest extends NLocalFileMetadataTestCase { Map<String, String> response = JsonUtil.readValueAsMap(result.getResponse().getContentAsString()); Assert.assertEquals(response.get("code"), KylinException.CODE_SUCCESS); - Mockito.verify(jobController).updateSparkJobTime(request); + Mockito.verify(jobController).updateSparkJobTime(Mockito.eq(request), Mockito.any()); + } + + @Test + public void testUpdateJobReportAcceptsServiceToken() throws Exception { + SecurityContextHolder.clearContext(); + ExecutablePO job = mockJob(ExecutableState.RUNNING); + JobErrorRequest request = new JobErrorRequest(); + request.setJobLastRunningStartTime(String.valueOf(job.getOutput().getLastRunningStartTime())); + request.setProject(job.getProject()); + request.setJobId(job.getId()); + request.setFailedStepId("c"); + request.setFailedSegmentId("d"); + request.setFailedStack("error"); + request.setFailedReason("reason"); + Mockito.doNothing().when(jobInfoService).updateJobError(request.getProject(), request.getJobId(), + request.getFailedStepId(), request.getFailedSegmentId(), request.getFailedStack(), + request.getFailedReason()); + + MvcResult result = mockMvc + .perform(MockMvcRequestBuilders.put("/api/jobs/error").contentType(MediaType.APPLICATION_JSON) + .header(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, + getTestConfig().getBroadcastToken()) + .content(JsonUtil.writeValueAsString(request)) + .accept(MediaType.parseMediaType(HTTP_VND_APACHE_KYLIN_JSON))) + .andExpect(MockMvcResultMatchers.status().isOk()).andReturn(); + Map<String, String> response = JsonUtil.readValueAsMap(result.getResponse().getContentAsString()); + Assert.assertEquals(response.get("code"), KylinException.CODE_SUCCESS); + + Mockito.verify(jobController).updateJobError(Mockito.eq(request), Mockito.any()); + } + + @Test + public void testUpdateJobReportRejectsAnonymous() { + SecurityContextHolder.clearContext(); + JobErrorRequest request = new JobErrorRequest(); + request.setProject("default"); + request.setJobId("job-id"); + Assert.assertThrows(KylinException.class, + () -> jobController.updateJobError(request, new MockHttpServletRequest())); } @Test diff --git a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/RestfulJobProgressReport.java b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/RestfulJobProgressReport.java index f09d7321d0..6a0b8a7f27 100644 --- a/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/RestfulJobProgressReport.java +++ b/src/spark-project/engine-spark/src/main/java/org/apache/kylin/engine/spark/job/RestfulJobProgressReport.java @@ -27,6 +27,7 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; +import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpHeaders; import org.apache.http.HttpResponse; import org.apache.http.HttpStatus; @@ -36,6 +37,7 @@ import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; import org.apache.kylin.common.KylinConfig; +import org.apache.kylin.common.persistence.transaction.BroadcastEventReadyNotifier; import org.apache.kylin.common.util.JsonUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -62,6 +64,10 @@ public class RestfulJobProgressReport implements IJobProgressReport { CloseableHttpClient httpClient = HttpClients.custom().setDefaultRequestConfig(defaultRequestConfig).build(); HttpPut httpPut = new HttpPut(requestApi); httpPut.addHeader(HttpHeaders.CONTENT_TYPE, HTTP_VND_APACHE_KYLIN_JSON); + String broadcastToken = KylinConfig.getInstanceFromEnv().getBroadcastToken(); + if (StringUtils.isNotBlank(broadcastToken)) { + httpPut.addHeader(BroadcastEventReadyNotifier.BROADCAST_TOKEN_HEADER, broadcastToken); + } httpPut.setEntity(new StringEntity(json, StandardCharsets.UTF_8)); HttpResponse response = httpClient.execute(httpPut);
