[INLONG-9671][Manager] Adding JDBC URL checks to the updateAuditSource method

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/AuditSourceType.java

@@ -0,0 +1,25 @@
+/*
+ * 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.common.consts;
+
+/**
+ * Audit Source Type
+ */
+public enum AuditSourceType {
+ MYSQL, CLICKHOUSE, ELASTICSEARCH
+}

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java

@@ -151,6 +151,10 @@ public enum ErrorCodeEnum {
 
  AUDIT_ID_TYPE_NOT_SUPPORTED(4001, "Audit id type '%s' not supported"),
 
+ AUDIT_SOURCE_TYPE_NOT_SUPPORTED(4002, "Audit Source type '%s' not supported"),
+
+ AUDIT_SOURCE_URL_NOT_SUPPORTED(4003, "Audit Source URL '%s' not supported"),
+
  TENANT_NOT_EXIST(5001, "Tenant '%s' is not exist"),
  ;
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/AbstractAuditSourceOperator.java

@@ -0,0 +1,29 @@
+/*
+ * 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.service.audit;
+
+/**
+ * Default of Audit Source Operator.
+ */
+public abstract class AbstractAuditSourceOperator implements InlongAuditSourceOperator {
+
+ @Override
+ public Boolean accept(String type) {
+ return getType().equals(type.toUpperCase());
+ }
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/AuditSourceClickhouseOperator.java

@@ -0,0 +1,48 @@
+/*
+ * 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.service.audit;
+
+import org.apache.inlong.manager.common.consts.AuditSourceType;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+/**
+ * Audit source clickhouse operator.
+ */
+@Service
+public class AuditSourceClickhouseOperator extends AbstractAuditSourceOperator {
+
+ private static final String CLICKHOUSE_URL_PREFIX = "jdbc:clickhouse://";
+
+ @Override
+ public String getType() {
+ return AuditSourceType.CLICKHOUSE.name();
+ }
+
+ @Override
+ public String convertTo(String url) {
+ if (StringUtils.isNotBlank(url) && url.startsWith(CLICKHOUSE_URL_PREFIX)) {
+ return url;
+ }
+
+ throw new BusinessException(String.format(ErrorCodeEnum.AUDIT_SOURCE_URL_NOT_SUPPORTED.getMessage(), url));
+ }
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/AuditSourceElasticsearchOperator.java

@@ -0,0 +1,50 @@
+/*
+ * 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.service.audit;
+
+import org.apache.inlong.manager.common.consts.AuditSourceType;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+/**
+ * Audit source elasticsearch operator.
+ */
+@Service
+public class AuditSourceElasticsearchOperator extends AbstractAuditSourceOperator {
+
+ private static final String ES_HTTP_HOST_PREFIX = "http://";
+ private static final String ES_HTTPS_HOST_PREFIX = "https://";
+
+ @Override
+ public String getType() {
+ return AuditSourceType.ELASTICSEARCH.name();
+ }
+
+ @Override
+ public String convertTo(String url) {
+ if (StringUtils.isNotBlank(url)
+ && (url.startsWith(ES_HTTP_HOST_PREFIX) || url.startsWith(ES_HTTPS_HOST_PREFIX))) {
+ return url;
+ }
+
+ throw new BusinessException(String.format(ErrorCodeEnum.AUDIT_SOURCE_URL_NOT_SUPPORTED.getMessage(), url));
+ }
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/AuditSourceMysqlOperator.java

@@ -0,0 +1,49 @@
+/*
+ * 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.service.audit;
+
+import org.apache.inlong.manager.common.consts.AuditSourceType;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.pojo.util.MySQLSensitiveUrlUtils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+/**
+ * Audit source mysql operator.
+ */
+@Service
+public class AuditSourceMysqlOperator extends AbstractAuditSourceOperator {
+
+ private static final String MYSQL_JDBC_PREFIX = "jdbc:mysql://";
+
+ @Override
+ public String getType() {
+ return AuditSourceType.MYSQL.name();
+ }
+
+ @Override
+ public String convertTo(String url) {
+ if (StringUtils.isNotBlank(url) && url.startsWith(MYSQL_JDBC_PREFIX)) {
+ return MySQLSensitiveUrlUtils.filterSensitive(url);
+ }
+
+ throw new BusinessException(String.format(ErrorCodeEnum.AUDIT_SOURCE_URL_NOT_SUPPORTED.getMessage(), url));
+ }
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/InlongAuditSourceOperator.java

@@ -0,0 +1,44 @@
+/*
+ * 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.service.audit;
+
+/**
+ * Interface of the inlong audit source operator.
+ */
+public interface InlongAuditSourceOperator {
+
+ /**
+ * Determines whether the current instance matches the specified type.
+ */
+ Boolean accept(String type);
+
+ /**
+ * Get the Audit Source type.
+ *
+ * @return audit source type string
+ */
+ String getType();
+
+ /**
+ * Convert the URL.
+ *
+ * @param url audit source url
+ * @return converted url string
+ */
+ String convertTo(String url);
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/audit/InlongAuditSourceOperatorFactory.java

@@ -0,0 +1,47 @@
+/*
+ * 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.service.audit;
+
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * Factory for {@link InlongAuditSourceOperator}.
+ */
+@Service
+public class InlongAuditSourceOperatorFactory {
+
+ @Autowired
+ private List<InlongAuditSourceOperator> auditSourceOperatorList;
+
+ /**
+ * Get an inlong Audit Source operator instance via the given type
+ */
+ public InlongAuditSourceOperator getInstance(String type) {
+ return auditSourceOperatorList.stream()
+ .filter(inst -> inst.accept(type))
+ .findFirst()
+ .orElseThrow(() -> new BusinessException(
+ String.format(ErrorCodeEnum.AUDIT_SOURCE_TYPE_NOT_SUPPORTED.getMessage(), type)));
+ }
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java

@@ -45,6 +45,8 @@
 import org.apache.inlong.manager.pojo.audit.AuditVO;
 import org.apache.inlong.manager.pojo.user.LoginUserUtils;
 import org.apache.inlong.manager.pojo.user.UserRoleCode;
+import org.apache.inlong.manager.service.audit.InlongAuditSourceOperator;
+import org.apache.inlong.manager.service.audit.InlongAuditSourceOperatorFactory;
 import org.apache.inlong.manager.service.core.AuditService;
 import org.apache.inlong.manager.service.resource.sink.ck.ClickHouseConfig;
 import org.apache.inlong.manager.service.resource.sink.es.ElasticsearchApi;
@@ -142,6 +144,8 @@ public class AuditServiceImpl implements AuditService {
  private AuditSourceEntityMapper auditSourceMapper;
  @Autowired
  private InlongGroupEntityMapper inlongGroupMapper;
+ @Autowired
+ private InlongAuditSourceOperatorFactory auditSourceOperatorFactory;
 
  @PostConstruct
  public void initialize() {
@@ -178,6 +182,9 @@ public Boolean refreshBaseItemCache() {
 
  @Override
  public Integer updateAuditSource(AuditSourceRequest request, String operator) {
+ InlongAuditSourceOperator auditSourceOperator = auditSourceOperatorFactory.getInstance(request.getType());
+ request.setUrl(auditSourceOperator.convertTo(request.getUrl()));
+
  String offlineUrl = request.getOfflineUrl();
  if (StringUtils.isNotBlank(offlineUrl)) {
  auditSourceMapper.offlineSourceByUrl(offlineUrl);