diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTO.java index b358cc9f4eb..91304ee9a31 100644 --- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTO.java +++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTO.java @@ -23,6 +23,7 @@ import org.apache.inlong.manager.common.util.CommonBeanUtils; import org.apache.inlong.manager.common.util.JsonUtils; import org.apache.inlong.manager.pojo.sink.BaseStreamSink; +import org.apache.inlong.manager.pojo.util.MySQLSensitiveUrlUtils; import com.google.common.base.Strings; import io.swagger.annotations.ApiModelProperty; @@ -87,6 +88,7 @@ public static OceanBaseSinkDTO getFromRequest(OceanBaseSinkRequest request, Stri OceanBaseSinkDTO dto = StringUtils.isNotBlank(extParams) ? OceanBaseSinkDTO.getFromJson(extParams) : new OceanBaseSinkDTO(); CommonBeanUtils.copyProperties(request, dto, true); + dto.setJdbcUrl(filterSensitive(request.getJdbcUrl())); return dto; } @@ -221,4 +223,17 @@ public static String setDbNameToUrlWithCdc(String jdbcUrl, String databaseName) } return resultUrl.toString(); } + + /** + * Filter sensitive params, see {@link MySQLSensitiveUrlUtils#filterSensitive(String)}. + * The OceanBase JDBC URL follows the same MySQL-protocol-compatible syntax as the MySQL sink, + * so the same sensitive-parameter filtering (e.g. autoDeserialize, allowLoadLocalInfile) applies. + * + * @param url str may have some sensitive params + * @return str without sensitive param + */ + public static String filterSensitive(String url) { + return MySQLSensitiveUrlUtils.filterSensitive(url); + } + } diff --git a/inlong-manager/manager-pojo/src/test/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTOTest.java b/inlong-manager/manager-pojo/src/test/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTOTest.java new file mode 100644 index 00000000000..b154e67931d --- /dev/null +++ b/inlong-manager/manager-pojo/src/test/java/org/apache/inlong/manager/pojo/sink/oceanbase/OceanBaseSinkDTOTest.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.inlong.manager.pojo.sink.oceanbase; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.net.URLEncoder; + +/** + * Test for {@link OceanBaseSinkDTO} + */ +public class OceanBaseSinkDTOTest { + + @Test + public void testFilterSensitive() throws Exception { + // the sensitive params no use url code + String originUrl = OceanBaseSinkDTO.filterSensitive( + "jdbc:oceanbase://127.0.0.1,(allowLoadLocalInfile=yeſ,allowUrlInLocalInfile=yeſ,allowLoadLocalInfileInPath=.,maxAllowedPacket=655360),:3307/test"); + Assertions.assertEquals( + "jdbc:oceanbase://127.0.0.1,(maxAllowedPacket=655360)(autoDeserialize=false,allowUrlInLocalInfile=false,allowLoadLocalInfile=false),:3307/test", + originUrl); + + originUrl = OceanBaseSinkDTO.filterSensitive( + "jdbc:oceanbase://127.0.0.1:3306?autoDeserialize=TRue&allowLoadLocalInfile = TRue&allowUrlInLocalInfile=TRue&allowLoadLocalInfileInPath=/&autoReconnect=true"); + Assertions.assertEquals( + "jdbc:oceanbase://127.0.0.1:3306?autoReconnect=true&autoDeserialize=false&allowUrlInLocalInfile=false&allowLoadLocalInfile=false", + originUrl); + + originUrl = OceanBaseSinkDTO.filterSensitive( + "jdbc:oceanbase://127.0.0.1:3306?autoDeserialize=Yes&allowLoadLocalInfile = Yes&autoReconnect=true&allowUrlInLocalInfile=YEs&allowLoadLocalInfileInPath=/"); + Assertions.assertEquals( + "jdbc:oceanbase://127.0.0.1:3306?autoReconnect=true&autoDeserialize=false&allowUrlInLocalInfile=false&allowLoadLocalInfile=false", + originUrl); + + // the sensitive params use url code + originUrl = OceanBaseSinkDTO.filterSensitive( + URLEncoder.encode( + "jdbc:oceanbase://127.0.0.1:3306?autoDeserialize=TRue&allowLoadLocalInfile = TRue&allowUrlInLocalInfile=TRue&allowLoadLocalInfileInPath=/&autoReconnect=true", + "UTF-8")); + Assertions.assertEquals( + "jdbc:oceanbase://127.0.0.1:3306?autoReconnect=true&autoDeserialize=false&allowUrlInLocalInfile=false&allowLoadLocalInfile=false", + originUrl); + } + + @Test + public void testGetFromRequestFiltersSensitiveParams() { + // Regression test for the OceanBase sink JDBC URL sensitive-parameter bypass: + // autoDeserialize=true must never survive getFromRequest(), since the resulting + // DTO's jdbcUrl is persisted and later forwarded verbatim to the Flink JDBC connector. + OceanBaseSinkRequest request = new OceanBaseSinkRequest(); + request.setJdbcUrl("jdbc:oceanbase://127.0.0.1:3306/test?autoDeserialize=true&allowLoadLocalInfile=true"); + + OceanBaseSinkDTO dto = OceanBaseSinkDTO.getFromRequest(request, null); + + Assertions.assertFalse(dto.getJdbcUrl().contains("autoDeserialize=true"), + "autoDeserialize=true must be neutralized before the OceanBase sink config is persisted"); + Assertions.assertTrue(dto.getJdbcUrl().contains("autoDeserialize=false")); + Assertions.assertTrue(dto.getJdbcUrl().contains("allowLoadLocalInfile=false")); + } + +}