Fix sink user pipe visibility and mask credentials in SHOW CREATE PIPE

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipePermissionIT.java

@@ -115,7 +115,7 @@ public void testSinkPermission() {
     try (final Connection connection = env.getConnection(BaseEnv.TABLE_SQL_DIALECT);
         final Statement statement = connection.createStatement()) {
       statement.execute(
-          "alter pipe a2b modify sink ('username'='thulab', 'password'='StrngPsWd@623451')");
+          "alter pipe a2b modify sink ('sink.username'='thulab', 'sink.password'='StrngPsWd@623451')");
     } catch (final SQLException e) {
       e.printStackTrace();
       fail("Alter pipe shall not fail if user and password are specified");
@@ -187,6 +187,13 @@ public void testSinkPermission() {
       final ResultSet result = statement.executeQuery("show pipes");
       Assert.assertTrue(result.next());
       Assert.assertFalse(result.next());
+
+      final ResultSet showCreateResult = statement.executeQuery("show create pipe a2b");
+      Assert.assertTrue(showCreateResult.next());
+      Assert.assertTrue(
+          showCreateResult.getString("Create Pipe").contains("'sink.password'='******'"));
+      Assert.assertFalse(showCreateResult.getString("Create Pipe").contains("StrngPsWd@623451"));
+      Assert.assertFalse(showCreateResult.next());
     } catch (Exception e) {
       fail(e.getMessage());
     }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/pipe/task/PipeTableResp.java

@@ -179,10 +179,10 @@ private boolean isVisible4SinkUser(final String userName, final PipeParameters s
     return Objects.equals(
         userName,
         sinkParameters.getStringByKeys(
-            PipeSourceConstant.EXTRACTOR_IOTDB_USER_KEY,
-            PipeSourceConstant.SOURCE_IOTDB_USER_KEY,
-            PipeSourceConstant.EXTRACTOR_IOTDB_USERNAME_KEY,
-            PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY));
+            PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY,
+            PipeSinkConstant.SINK_IOTDB_USER_KEY,
+            PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY,
+            PipeSinkConstant.SINK_IOTDB_USERNAME_KEY));
   }
 
   public TGetAllPipeInfoResp convertToTGetAllPipeInfoResp() throws IOException {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowCreatePipeTask.java

@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
 import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
 import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import com.google.common.util.concurrent.ListenableFuture;
@@ -47,6 +48,8 @@
 
 public class ShowCreatePipeTask implements IConfigTask {
 
+  private static final String HIDDEN_VALUE = "******";
+
   private final String pipeName;
   private final String userName;
 
@@ -167,10 +170,12 @@ private static void appendAttributesClause(
     }
     final List<String> pairs = new ArrayList<>(attributes.size());
     for (final Map.Entry<String, String> entry : attributes.entrySet()) {
+      final String value =
+          PipeParameters.ValueHider.isHiddenKey(entry.getKey()) ? HIDDEN_VALUE : entry.getValue();
       pairs.add(
           ShowCreateTableTask.getString(entry.getKey())
               + "="
-              + ShowCreateTableTask.getString(entry.getValue()));
+              + ShowCreateTableTask.getString(value));
     }
     builder.append(" ").append(clause).append(" (").append(String.join(",", pairs)).append(")");
   }

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/relational/ShowCreateTaskTest.java

@@ -39,6 +39,7 @@
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 
 public class ShowCreateTaskTest {
 
@@ -114,7 +115,7 @@ public void testShowCreatePipeSQLShouldSanitizeInternalAndInjectedAttributes() {
   }
 
   @Test
-  public void testShowCreatePipeSQLShouldKeepExplicitCredentials() {
+  public void testShowCreatePipeSQLShouldMaskExplicitCredentials() {
     final Map<String, String> sourceAttributes = new HashMap<>();
     sourceAttributes.put(PipeSourceConstant.SOURCE_KEY, "iotdb-source");
     sourceAttributes.put(PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY, "alice");
@@ -136,13 +137,13 @@ public void testShowCreatePipeSQLShouldKeepExplicitCredentials() {
 
     assertEquals(
         "CREATE PIPE \"test_pipe\""
-            + " WITH SOURCE ('source'='iotdb-source','source.password'='secret','source.username'='alice')"
-            + " WITH SINK ('sink'='write-back-sink','sink.password'='secret','sink.username'='alice')",
+            + " WITH SOURCE ('source'='iotdb-source','source.password'='******','source.username'='alice')"
+            + " WITH SINK ('sink'='write-back-sink','sink.password'='******','sink.username'='alice')",
         ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta));
   }
 
   @Test
-  public void testShowCreatePipeSQLShouldKeepExplicitCredentialsWhenInjectionMarkerIsReset() {
+  public void testShowCreatePipeSQLShouldMaskExplicitCredentialsWhenInjectionMarkerIsReset() {
     final Map<String, String> sourceAttributes = new HashMap<>();
     sourceAttributes.put(PipeSourceConstant.SOURCE_KEY, "iotdb-source");
     sourceAttributes.put(PipeSourceConstant.SOURCE_IOTDB_USERNAME_KEY, "alice");
@@ -170,11 +171,44 @@ public void testShowCreatePipeSQLShouldKeepExplicitCredentialsWhenInjectionMarke
 
     assertEquals(
         "CREATE PIPE \"test_pipe\""
-            + " WITH SOURCE ('source'='iotdb-source','source.password'='secret','source.username'='alice')"
-            + " WITH SINK ('sink'='write-back-sink','sink.password'='secret','sink.username'='alice')",
+            + " WITH SOURCE ('source'='iotdb-source','source.password'='******','source.username'='alice')"
+            + " WITH SINK ('sink'='write-back-sink','sink.password'='******','sink.username'='alice')",
         ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta));
   }
 
+  @Test
+  public void testShowCreatePipeSQLShouldMaskSensitiveAliasAttributes() {
+    final Map<String, String> sourceAttributes = new HashMap<>();
+    sourceAttributes.put(PipeSourceConstant.EXTRACTOR_KEY, "iotdb-extractor");
+    sourceAttributes.put(PipeSourceConstant.EXTRACTOR_IOTDB_PASSWORD_KEY, "source-secret");
+    sourceAttributes.put("extractor.ssl.key-store-pwd", "key-store-secret");
+
+    final Map<String, String> sinkAttributes = new HashMap<>();
+    sinkAttributes.put(PipeSinkConstant.CONNECTOR_KEY, "iotdb-thrift-connector");
+    sinkAttributes.put(PipeSinkConstant.CONNECTOR_IOTDB_PASSWORD_KEY, "sink-secret");
+
+    final PipeMeta pipeMeta =
+        new PipeMeta(
+            new PipeStaticMeta("test_pipe", 1L, sourceAttributes, new HashMap<>(), sinkAttributes),
+            new PipeRuntimeMeta());
+
+    final String sql = ShowCreatePipeTask.getShowCreatePipeSQL(pipeMeta);
+    assertEquals(3, countOccurrences(sql, "******"));
+    assertFalse(sql.contains("source-secret"));
+    assertFalse(sql.contains("key-store-secret"));
+    assertFalse(sql.contains("sink-secret"));
+  }
+
+  private static int countOccurrences(final String value, final String searchedValue) {
+    int count = 0;
+    int index = 0;
+    while ((index = value.indexOf(searchedValue, index)) >= 0) {
+      count++;
+      index += searchedValue.length();
+    }
+    return count;
+  }
+
   @Test
   public void testShowCreatePipeSQLShouldSanitizeExtractorAndConnectorAliases() {
     final Map<String, String> sourceAttributes = new HashMap<>();