diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java index add7761ddc2..a1ee8ccc85a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java @@ -79,6 +79,7 @@ import org.apache.doris.qe.ShowResultSetMetaData; import org.apache.doris.rpc.RpcException; import org.apache.doris.service.FrontendOptions; import org.apache.doris.system.Backend; +import org.apache.doris.tablefunction.CdcStreamTableValuedFunction; import org.apache.doris.tablefunction.S3TableValuedFunction; import org.apache.doris.thrift.TCell; import org.apache.doris.thrift.TRow; @@ -267,6 +268,8 @@ public class StreamingInsertJob extends AbstractJob createTbls = createTableIfNotExists(); @@ -429,9 +432,13 @@ public class StreamingInsertJob extends AbstractJob properties) { JdbcClientConfig config = new JdbcClientConfig(); config.setCatalog(sourceType.name()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java index efa1757a6d5..965989002f6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java @@ -28,6 +28,7 @@ import org.apache.doris.job.common.JobStatus; import org.apache.doris.job.extensions.insert.streaming.DataSourceConfigValidator; import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob; import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties; +import org.apache.doris.job.util.StreamingJobUtils; import org.apache.doris.nereids.analyzer.UnboundTVFRelation; import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.trees.plans.PlanType; @@ -168,6 +169,8 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne boolean sourcePropModified = isPropertiesModified(streamingJob.getSourceProperties(), this.getSourceProperties()); if (sourcePropModified) { + StreamingJobUtils.checkCdcDriverPrivilege(ConnectContext.get(), + sourceProperties.get(DataSourceConfigKeys.DRIVER_URL)); DataSourceConfigValidator.validateSource(this.getSourceProperties(), streamingJob.getDataSourceType().name()); checkUnmodifiableSourceProperties(streamingJob.getSourceProperties()); @@ -325,7 +328,8 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne inputTvf.getProperties().getMap().get(unmodifiable)), "The '%s' property cannot be modified in ALTER JOB for cdc_stream", unmodifiable); } - CdcStreamTableValuedFunction.validateProperties(inputTvf.getProperties().getMap()); + // Validate merged resource properties as the ALTER caller, before schema discovery. + new CdcStreamTableValuedFunction(inputTvf.getProperties().getMap()).checkAuth(ConnectContext.get()); break; default: throw new IllegalArgumentException("Unsupported tvf type:" + inputTvf.getFunctionName()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java index 0a53ec38479..14f02c80a62 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java @@ -29,6 +29,7 @@ import org.apache.doris.job.common.DataSourceType; import org.apache.doris.job.extensions.insert.streaming.DataSourceConfigValidator; import org.apache.doris.job.extensions.insert.streaming.StreamingJdbcUrlNormalizer; import org.apache.doris.job.util.StreamingJobUtils; +import org.apache.doris.qe.ConnectContext; import org.apache.doris.thrift.TBrokerFileStatus; import org.apache.doris.thrift.TFileType; @@ -83,6 +84,8 @@ public class CdcStreamTableValuedFunction extends ExternalFileTableValuedFunctio StreamingJobUtils.populateDefaultSourceProperties(sourceType, copyProps, jobId); super.parseCommonProperties(copyProps); + StreamingJobUtils.checkCdcDriverPrivilege(ConnectContext.get(), + processedParams.get(DataSourceConfigKeys.DRIVER_URL)); this.processedParams.put(ENABLE_CDC_CLIENT_KEY, "true"); this.processedParams.put(URI_KEY, URI); this.processedParams.put(HTTP_ENABLE_RANGE_REQUEST_KEY, "false"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java index 63bce5df853..c07703f4b2c 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java @@ -17,10 +17,13 @@ package org.apache.doris.job.extensions.insert.streaming; +import org.apache.doris.analysis.UserIdentity; import org.apache.doris.catalog.Env; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.Config; import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.job.cdc.DataSourceConfigKeys; import org.apache.doris.job.cdc.request.CommitOffsetRequest; import org.apache.doris.job.cdc.split.SnapshotSplit; @@ -34,8 +37,10 @@ import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider; import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider; import org.apache.doris.job.offset.s3.S3Offset; import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; +import org.apache.doris.nereids.exceptions.ParseException; import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand; import org.apache.doris.persist.gson.GsonUtils; +import org.apache.doris.qe.ConnectContext; import org.apache.doris.tablefunction.CdcStreamTableValuedFunction; import org.apache.doris.tablefunction.S3TableValuedFunction; import org.apache.doris.transaction.GlobalTransactionMgrIface; @@ -121,21 +126,49 @@ public class StreamingInsertJobOffsetPersistenceTest { Deencapsulation.setField(job, "jobId", 9001L); job.setJobStatus(JobStatus.PAUSED); Deencapsulation.setField(job, "tvfType", CdcStreamTableValuedFunction.NAME); - HashMap tvfProperties = new HashMap<>(); - tvfProperties.put(DataSourceConfigKeys.TYPE, "postgres"); - tvfProperties.put(DataSourceConfigKeys.JDBC_URL, "jdbc:postgresql://127.0.0.1:5432/source_db"); - tvfProperties.put(DataSourceConfigKeys.TABLE, "source_table"); - Deencapsulation.setField(job, "originTvfProps", tvfProperties); + Deencapsulation.setField(job, "createUser", UserIdentity.ROOT); + job.setExecuteSql("INSERT INTO internal.test_db.target_table SELECT * FROM cdc_stream(" + + "\"type\"=\"postgres\", \"jdbc_url\"=\"jdbc:postgresql://127.0.0.1:5432/source_db\", " + + "\"table\"=\"source_table\")"); job.gsonPostProcess(); JdbcTvfSourceOffsetProvider provider = Deencapsulation.getField(job, "offsetProvider"); Assertions.assertNull(Deencapsulation.getField(provider, "sourceType")); - job.validateAlterOffset("{\"lsn\":\"300\"}"); + ConnectContext previousContext = ConnectContext.get(); + try (MockedStatic envMockedStatic = Mockito.mockStatic(Env.class)) { + Env env = Mockito.mock(Env.class); + InternalCatalog catalog = Mockito.mock(InternalCatalog.class); + CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class); + Mockito.when(catalog.getName()).thenReturn(InternalCatalog.INTERNAL_CATALOG_NAME); + Mockito.when(env.getInternalCatalog()).thenReturn(catalog); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr); + Mockito.when(catalogMgr.getCatalog(InternalCatalog.INTERNAL_CATALOG_NAME)).thenReturn(catalog); + envMockedStatic.when(Env::getCurrentEnv).thenReturn(env); + ConnectContext caller = new ConnectContext(); + caller.setCurrentUserIdentity(UserIdentity.createAnalyzedUserIdentWithIp("alter_user", "%")); + caller.setAuthenticatedRoles(Collections.singleton("session_role")); + caller.setThreadLocalInfo(); - Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); - Assertions.assertThrows(AnalysisException.class, - () -> job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}")); + // Restored jobs have no cached plan/TVF properties. Parsing must not replace the caller. + job.validateAlterOffset("{\"lsn\":\"300\"}"); + + Assertions.assertSame(caller, ConnectContext.get()); + Assertions.assertEquals(Collections.singleton("session_role"), + ConnectContext.get().getAuthenticatedRoles()); + Assertions.assertEquals(JobStatus.PAUSED, job.getJobStatus()); + Assertions.assertThrows(AnalysisException.class, + () -> job.validateAlterOffset("{\"file\":\"mysql-bin.000001\",\"pos\":\"154\"}")); + + job.setExecuteSql("INSERT INTO"); + Assertions.assertThrows(ParseException.class, () -> job.initLogicalPlan(true)); + Assertions.assertSame(caller, ConnectContext.get()); + } finally { + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } } @Test diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java index b55fa2c817c..ee72dd9c043 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/util/StreamingJobUtilsTest.java @@ -30,6 +30,10 @@ import org.apache.doris.datasource.jdbc.client.JdbcClient; import org.apache.doris.job.cdc.DataSourceConfigKeys; import org.apache.doris.job.common.DataSourceType; import org.apache.doris.job.exception.JobException; +import org.apache.doris.mysql.privilege.AccessControllerManager; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.GlobalVariable; import org.junit.jupiter.api.Assertions; @@ -57,6 +61,32 @@ public class StreamingJobUtilsTest { MockitoAnnotations.initMocks(this); } + @Test + public void testCdcDriverPrivilege() { + Env env = Mockito.mock(Env.class); + ConnectContext ctx = Mockito.mock(ConnectContext.class); + AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class); + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + Mockito.when(env.getAccessManager()).thenReturn(accessManager); + List remoteDrivers = List.of("http://example.invalid/driver.jar", + "https://example.invalid/driver.jar"); + for (String driver : remoteDrivers) { + AnalysisException error = Assertions.assertThrows(AnalysisException.class, + () -> StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver)); + Assertions.assertTrue(error.getMessage().contains("ADMIN")); + } + Mockito.when(accessManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(true); + for (String driver : remoteDrivers) { + Assertions.assertDoesNotThrow(() -> StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver)); + } + Mockito.when(accessManager.checkGlobalPriv(ctx, PrivPredicate.ADMIN)).thenReturn(false); + for (String driver : Arrays.asList(null, "driver.jar", "file:///opt/doris/driver.jar")) { + Assertions.assertDoesNotThrow(() -> StreamingJobUtils.checkCdcDriverPrivilege(ctx, driver)); + } + } + } + @Test public void testBinarySourceColumnsUseSupportedOlapTypes() { List columns = new ArrayList<>(); diff --git a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy index 6100c7eee9e..1507d3c3b5f 100644 --- a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy +++ b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_priv.groovy @@ -100,14 +100,34 @@ suite("test_streaming_mysql_job_priv", "p0,external,mysql,external_docker,extern sql """INSERT INTO ${mysqlDb}.${tableName} (name, age) VALUES ('B1', 2);""" } + // Prepare a local driver without granting ADMIN to the job creator. + File driverFile = new File("${context.config.dataPath}/jdbc_driver/mysql_priv_driver.jar").absoluteFile + driverFile.parentFile.mkdirs() + def download = new ProcessBuilder("curl", "--fail", "--location", "--max-time", "120", + driver_url, "--output", driverFile.path).redirectErrorStream(true).start() + String downloadOutput = download.text + assertEquals(0, download.waitFor(), downloadOutput) + scp_udf_file_to_all_fe(driverFile.path) + String localDriverUrl = "file://${driverFile.path}" + // Create the job without ALTER privilege to verify schema changes use the job creator's identity. sql """grant load_priv,create_priv on ${dbName}.* to ${user}""" connect(user, "${pwd}", url) { + test { + sql """CREATE JOB ${jobName} ON STREAMING + FROM MYSQL ( + "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}", + "driver_url" = "${driver_url}", "driver_class" = "com.mysql.cj.jdbc.Driver", + "user" = "root", "password" = "123456", "database" = "${mysqlDb}", + "include_tables" = "${tableName}", "offset" = "initial") + TO DATABASE ${currentDb} ("table.create.properties.replication_num" = "1")""" + exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege" + } sql """CREATE JOB ${jobName} ON STREAMING FROM MYSQL ( "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}?allowPublicKeyRetrieval=true&useSSL=false", - "driver_url" = "${driver_url}", + "driver_url" = "${localDriverUrl}", "driver_class" = "com.mysql.cj.jdbc.Driver", "user" = "root", "password" = "123456", @@ -191,6 +211,13 @@ suite("test_streaming_mysql_job_priv", "p0,external,mysql,external_docker,extern } sql "PAUSE JOB where jobname = '${jobName}'"; + connect(user, "${pwd}", url) { + test { + sql """ALTER JOB ${jobName} FROM MYSQL ("driver_url" = "${driver_url}") + TO DATABASE ${currentDb}""" + exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege" + } + } sql """ALTER JOB ${jobName} FROM MYSQL ( diff --git a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy index 744ab7539ac..c9b6b766e82 100644 --- a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy +++ b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy @@ -138,6 +138,27 @@ suite("test_cdc_stream_tvf_mysql", "p0,external,mysql,external_docker,external_d exception "ssl_mode 'verify-ca' requires ssl_rootcert to be set" } + // Remote driver URLs require ADMIN even for standalone TVF queries. + String driverUser = "test_cdc_stream_tvf_mysql_driver_user" + def jdbcTokens = context.getJdbcUrl().split('/') + String driverUserUrl = jdbcTokens[0] + "//" + jdbcTokens[2] + "/" + currentDb + "?" + sql """DROP USER IF EXISTS '${driverUser}'""" + sql """CREATE USER '${driverUser}' IDENTIFIED BY '123456'""" + sql """GRANT SELECT_PRIV ON ${currentDb}.* TO '${driverUser}'""" + try { + connectToDoris(driverUser, "123456", driverUserUrl) { + test { + sql """SELECT * FROM cdc_stream( + "type" = "mysql", "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}", + "database" = "${mysqlDb}", "table" = "${table1}", "offset" = "latest", + "driver_url" = "${driver_url}")""" + exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege" + } + } + } finally { + sql """DROP USER '${driverUser}'""" + } + // --- Data setup --- connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysql_port}") { diff --git a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy index d0848257a8b..78d8b76a897 100644 --- a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy +++ b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql.groovy @@ -67,10 +67,8 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker, sql """INSERT INTO ${mysqlDb}.${mysqlTable} (name, age) VALUES ('B1', 2)""" } - // create streaming job via cdc_stream TVF (offset=initial → snapshot then binlog) - sql """ - CREATE JOB ${jobName} - ON STREAMING DO INSERT INTO ${currentDb}.${dorisTable} (name, age) + String insertSql = """ + INSERT INTO ${currentDb}.${dorisTable} (name, age) SELECT name, age FROM cdc_stream( "type" = "mysql", "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}", @@ -85,6 +83,25 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker, ) """ + String driverUser = "test_streaming_cdc_mysql_driver_user" + def jdbcTokens = context.getJdbcUrl().split('/') + String driverUserUrl = jdbcTokens[0] + "//" + jdbcTokens[2] + "/" + currentDb + "?" + sql """DROP USER IF EXISTS '${driverUser}'""" + sql """CREATE USER '${driverUser}' IDENTIFIED BY '123456'""" + sql """GRANT LOAD_PRIV ON ${currentDb}.* TO '${driverUser}'""" + if (isCloudMode()) { + def cluster = sql_return_maparray("SHOW CLUSTERS").find { it.is_current.equalsIgnoreCase("TRUE") }.cluster + sql """GRANT USAGE_PRIV ON CLUSTER `${cluster}` TO '${driverUser}'""" + } + connectToDoris(driverUser, "123456", driverUserUrl) { + test { + sql """CREATE JOB ${jobName} ON STREAMING DO ${insertSql}""" + exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege" + } + } + // The administrator can create the same job with the remote driver. + sql """CREATE JOB ${jobName} ON STREAMING DO ${insertSql}""" + // wait for at least one snapshot task to succeed try { Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({ @@ -130,6 +147,33 @@ suite("test_streaming_job_cdc_stream_mysql", "p0,external,mysql,external_docker, s.size() == 1 && s.get(0).get(0) == "PAUSED" }) + String driverResource = "test_streaming_cdc_mysql_driver_resource" + try_sql """DROP RESOURCE '${driverResource}'""" + try { + sql """CREATE RESOURCE '${driverResource}' PROPERTIES ( + "type" = "jdbc", + "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}", + "user" = "root", "password" = "123456", + "driver_url" = "${driver_url}", "driver_class" = "com.mysql.cj.jdbc.Driver" + )""" + // ALTER does not read the local jar; the conflicting SQL must be rejected first. + sql """ALTER RESOURCE '${driverResource}' PROPERTIES ("driver_url" = "file:///trusted/mysql.jar")""" + sql """GRANT USAGE_PRIV ON RESOURCE '${driverResource}' TO '${driverUser}'""" + String inheritedDriverSql = insertSql.replace('"driver_url"', + '"resource" = "' + driverResource + '", "Driver_url"') + .replace(driver_url, 'https://example.invalid/driver.jar') + .replace('com.mysql.cj.jdbc.Driver', 'example.InvalidDriver') + connectToDoris(driverUser, "123456", driverUserUrl) { + test { + sql """ALTER JOB ${jobName} ${inheritedDriverSql}""" + exception "HTTP/HTTPS CDC driver URLs require ADMIN privilege" + } + } + } finally { + sql """DROP RESOURCE '${driverResource}'""" + } + sql """DROP USER '${driverUser}'""" + test { sql """ ALTER JOB ${jobName}