[Fix](auth) Restrict processlist visibility by user identity

be/src/information_schema/schema_processlist_scanner.cpp

@@ -58,6 +58,7 @@ Status SchemaProcessListScanner::start(RuntimeState* state) {
     TShowProcessListRequest request;
     request.__set_show_full_sql(true);
     request.__set_time_zone(state->timezone());
+    request.__set_current_user_ident(*_param->common_param->current_user_ident);
 
     for (const auto& fe_addr : _param->common_param->fe_addr_list) {
         TShowProcessListResult tmp_ret;

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -5605,15 +5605,15 @@ public TStatus reportCommitTxnResult(TReportCommitTxnResultRequest request) thro
     }
 
     @Override
-    public TShowProcessListResult showProcessList(TShowProcessListRequest request) {
+    public TShowProcessListResult showProcessList(TShowProcessListRequest request) throws TException {
+        if (!request.isSetCurrentUserIdent()) {
+            throw new TException("Current user identity is not set");
+        }
         boolean isShowFullSql = false;
         if (request.isSetShowFullSql()) {
             isShowFullSql = request.isShowFullSql();
         }
-        UserIdentity userIdentity = UserIdentity.ROOT;
-        if (request.isSetCurrentUserIdent()) {
-            userIdentity = UserIdentity.fromThrift(request.getCurrentUserIdent());
-        }
+        UserIdentity userIdentity = UserIdentity.fromThrift(request.getCurrentUserIdent());
         String timeZone = VariableMgr.getDefaultSessionVariable().getTimeZone();
         if (request.isSetTimeZone()) {
             timeZone = request.getTimeZone();

fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java

@@ -56,6 +56,7 @@
 import org.apache.doris.thrift.TRollbackTxnRequest;
 import org.apache.doris.thrift.TSchemaTableName;
 import org.apache.doris.thrift.TSchemaTableRequestParams;
+import org.apache.doris.thrift.TShowProcessListRequest;
 import org.apache.doris.thrift.TShowUserRequest;
 import org.apache.doris.thrift.TShowUserResult;
 import org.apache.doris.thrift.TStatusCode;
@@ -66,6 +67,7 @@
 import org.apache.doris.utframe.TestWithFeService;
 
 import com.google.common.collect.Sets;
+import org.apache.thrift.TException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
@@ -117,6 +119,15 @@ private static void setPrivateField(Object target, String fieldName, Object valu
         field.set(target, value);
     }
 
+    @Test
+    public void testShowProcessListRejectsMissingUserIdentity() {
+        FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+        TShowProcessListRequest request = new TShowProcessListRequest();
+
+        TException exception = Assertions.assertThrows(TException.class, () -> impl.showProcessList(request));
+        Assertions.assertEquals("Current user identity is not set", exception.getMessage());
+    }
+
     @Test
     public void testGetTableNamesWithSysTablePattern() throws Exception {
         FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);

regression-test/suites/show_p0/test_show_processlist.groovy

@@ -18,6 +18,20 @@
 import org.apache.doris.regression.util.Http
 
 suite("test_show_processlist") {
+    def victimUser = "test_processlist_victim"
+    def attackerUser = "test_processlist_attacker"
+    def userPassword = "C123_567p"
+    try_sql "DROP USER '${victimUser}'"
+    try_sql "DROP USER '${attackerUser}'"
+    sql "CREATE USER '${victimUser}' IDENTIFIED BY '${userPassword}'"
+    sql "CREATE USER '${attackerUser}' IDENTIFIED BY '${userPassword}'"
+    sql "GRANT SELECT_PRIV ON regression_test.* TO '${victimUser}'"
+    sql "GRANT SELECT_PRIV ON regression_test.* TO '${attackerUser}'"
+    if (isCloudMode()) {
+        sql "GRANT USAGE_PRIV ON COMPUTE GROUP '%' TO '${victimUser}'"
+        sql "GRANT USAGE_PRIV ON COMPUTE GROUP '%' TO '${attackerUser}'"
+    }
+
     sql """set fetch_all_fe_for_system_table = false;"""
     def result = sql """show processlist;"""
     logger.info("result:${result}")
@@ -42,7 +56,39 @@ suite("test_show_processlist") {
     logger.info("result:${result}")
     assertTrue(result[0].size() == 15)
 
-
+    connect(victimUser, userPassword, context.config.jdbcUrl) {
+        sql "select 1"
+        connect(attackerUser, userPassword, context.config.jdbcUrl) {
+            def attackerRows = sql """
+                SELECT User, Info
+                FROM information_schema.processlist
+                WHERE User IN ('${victimUser}', '${attackerUser}')
+                ORDER BY User
+            """
+            assertFalse(attackerRows.isEmpty())
+            assertTrue(attackerRows.every { row -> row[0] == attackerUser })
+            assertFalse(attackerRows.any { row -> row[0] == victimUser })
+            assertTrue(attackerRows.any { row ->
+                row[1] != null && row[1].toString().contains("information_schema.processlist")
+            })
+
+            def showRows = sql "SHOW FULL PROCESSLIST"
+            assertFalse(showRows.isEmpty())
+            assertTrue(showRows.every { row -> row[2] == attackerUser })
+
+            connect('root', context.config.jdbcPassword, context.config.jdbcUrl) {
+                def adminRows = sql """
+                    SELECT User
+                    FROM information_schema.processlist
+                    WHERE User IN ('${victimUser}', '${attackerUser}')
+                    ORDER BY User
+                """
+                assertTrue(adminRows.any { row -> row[0] == victimUser })
+                assertTrue(adminRows.any { row -> row[0] == attackerUser })
+            }
+        }
+    }
+
     def result1 = connect('root', context.config.jdbcPassword, context.config.jdbcUrl) {
         // execute sql with admin user
         sql 'select 99 + 1'