[fix](fe) Preserve session roles in metadata authorization

fe/fe-core/src/main/java/org/apache/doris/analysis/UserIdentity.java

@@ -40,7 +40,9 @@
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 // https://dev.mysql.com/doc/refman/8.0/en/account-names.html
 // user name must be literally matched.
@@ -68,6 +70,9 @@ public class UserIdentity implements Writable, GsonPostProcessable {
     @SerializedName(value = "subject")
     private String subject;
 
+    // Request-scoped roles carried by trusted internal RPCs. Never persist these as account grants.
+    private transient Set<String> authenticatedRoles = Collections.emptySet();
+
     private boolean isAnalyzed = false;
 
     public static final UserIdentity ROOT;
@@ -111,12 +116,25 @@ public static UserIdentity createAnalyzedUserIdentWithDomain(String user, String
     }
 
     public static UserIdentity fromThrift(TUserIdentity tUserIdent) {
-        UserIdentity userIdentity = new UserIdentity(tUserIdent.getUsername(),
-                tUserIdent.getHost(), tUserIdent.is_domain);
+        return createAnalyzedUserIdentWithRoles(tUserIdent.getUsername(), tUserIdent.getHost(), tUserIdent.is_domain,
+                tUserIdent.isSetAuthenticatedRoles() ? tUserIdent.getAuthenticatedRoles() : Collections.emptySet());
+    }
+
+    /** Construct a request-scoped authorization identity without modifying the stored account. */
+    public static UserIdentity createAnalyzedUserIdentWithRoles(String user, String host, boolean isDomain,
+            Set<String> authenticatedRoles) {
+        UserIdentity userIdentity = new UserIdentity(user, host, isDomain);
         userIdentity.setIsAnalyzed();
+        if (!authenticatedRoles.isEmpty()) {
+            userIdentity.authenticatedRoles = Collections.unmodifiableSet(new HashSet<>(authenticatedRoles));
+        }
         return userIdentity;
     }
 
+    public Set<String> getAuthenticatedRoles() {
+        return authenticatedRoles;
+    }
+
     public String getQualifiedUser() {
         Preconditions.checkState(isAnalyzed);
         return user;
@@ -298,11 +316,18 @@ public boolean isSystemUser() {
     }
 
     public TUserIdentity toThrift() {
+        return toThrift(authenticatedRoles);
+    }
+
+    public TUserIdentity toThrift(Set<String> sessionRoles) {
         Preconditions.checkState(isAnalyzed);
         TUserIdentity tUserIdent = new TUserIdentity();
         tUserIdent.setHost(host);
         tUserIdent.setUsername(user);
         tUserIdent.setIsDomain(isDomain);
+        if (!sessionRoles.isEmpty()) {
+            tUserIdent.setAuthenticatedRoles(new HashSet<>(sessionRoles));
+        }
         return tUserIdent;
     }
 

fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java

@@ -214,7 +214,7 @@ private TGetDbsResult getDbNames() throws TException {
         if (dbSchemaFilterPattern != null) {
             getDbsParams.setPattern(dbSchemaFilterPattern);
         }
-        getDbsParams.setCurrentUserIdent(ctx.getCurrentUserIdentity().toThrift());
+        getDbsParams.setCurrentUserIdent(ctx.getCurrentUserIdentityForRpc());
         return impl.getDbNames(getDbsParams);
     }
 
@@ -232,7 +232,7 @@ private TListTableStatusResult listTableStatus(String dbName, String catalogName
         }
         // Flight SQL only consumes names and types, not the potentially expensive table statistics.
         getTablesParams.setRequiredColumns(Set.of("TABLE_NAME", "TABLE_TYPE"));
-        getTablesParams.setCurrentUserIdent(ctx.getCurrentUserIdentity().toThrift());
+        getTablesParams.setCurrentUserIdent(ctx.getCurrentUserIdentityForRpc());
         return impl.listTableStatus(getTablesParams);
     }
 
@@ -247,7 +247,7 @@ private TDescribeTablesResult describeTables(String dbName, String catalogName,
             describeTablesParams.setCatalog(catalogName);
         }
         describeTablesParams.setTablesName(tablesName);
-        describeTablesParams.setCurrentUserIdent(ctx.getCurrentUserIdentity().toThrift());
+        describeTablesParams.setCurrentUserIdent(ctx.getCurrentUserIdentityForRpc());
         return impl.describeTables(describeTablesParams);
     }
 

fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/MetadataScanNode.java

@@ -63,7 +63,7 @@ protected void toThrift(TPlanNode planNode) {
         TMetaScanNode metaScanNode = new TMetaScanNode();
         metaScanNode.setTupleId(desc.getId().asInt());
         metaScanNode.setMetadataType(this.tvf.getMetadataType());
-        TUserIdentity tCurrentUser = ConnectContext.get().getCurrentUserIdentity().toThrift();
+        TUserIdentity tCurrentUser = ConnectContext.get().getCurrentUserIdentityForRpc();
         metaScanNode.setCurrentUserIdent(tCurrentUser);
         planNode.setMetaScanNode(metaScanNode);
     }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java

@@ -343,7 +343,7 @@ private AuthorizationPlugin adapt(String name, CatalogAccessController controlle
         // the check that provoked it, because the source answering for instance scope may well be a plugin
         // that looks at the client address, and this question is about the same statement.
         return new LegacyAccessControllerPlugin(name, controller,
-                (subject, requirement, context) -> decide(AccessTranslation.userIdentityOf(subject),
+                (subject, requirement, context) -> decide(AccessTranslation.userIdentityOf(subject, context),
                         AuthorizedResource.global(), requirement, context));
     }
 
@@ -1019,10 +1019,11 @@ public void decideColumns(UserIdentity subject, AuthorizedResource.Columns colum
     private void ask(UserIdentity subject, AuthorizedResource resource, AccessRequirement requirement,
             AccessContext context) throws AccessDeniedException {
         AuthorizationPlugin controller = controllerOf(resource);
-        AccessContext outer = enterCheck(context);
+        AccessContext identityContext = UserIdentityAccessContext.of(subject, context);
+        AccessContext outer = enterCheck(identityContext);
         try {
             checkInClassLoaderOf(controller, () -> controller.checkPrivilege(
-                    AccessTranslation.subjectOf(subject), resource, requirement, context));
+                    AccessTranslation.subjectOf(subject), resource, requirement, identityContext));
         } finally {
             leaveCheck(outer);
         }
@@ -1324,7 +1325,7 @@ public Map<String, DataMaskSpec> evalDataMaskPolicies(UserIdentity currentUser,
         }
         AuthorizedResource.Table table = AuthorizedResource.table(ctl, db, tbl);
         AuthorizationPlugin controller = controllerOf(table);
-        AccessContext context = ConnectionAccessContext.current();
+        AccessContext context = UserIdentityAccessContext.of(currentUser, ConnectionAccessContext.current());
         AccessContext outer = enterCheck(context);
         try {
             return answered(controller, "the data masks of " + table,
@@ -1343,7 +1344,7 @@ public List<RowFilterSpec> evalRowFilterPolicies(UserIdentity currentUser, Strin
         Objects.requireNonNull(tbl, "require tbl object");
         AuthorizedResource.Table table = AuthorizedResource.table(ctl, db, tbl);
         AuthorizationPlugin controller = controllerOf(table);
-        AccessContext context = ConnectionAccessContext.current();
+        AccessContext context = UserIdentityAccessContext.of(currentUser, ConnectionAccessContext.current());
         AccessContext outer = enterCheck(context);
         try {
             return answered(controller, "the row filters of " + table,
@@ -1362,7 +1363,8 @@ public DataPolicyContext captureDataPolicyContext(UserIdentity user, String ctl,
         }
         AuthorizedResource.Table table = AuthorizedResource.table(ctl, db, tbl);
         return new DataPolicyContext(controllerOf(table), AccessTranslation.subjectOf(user), table,
-                columns, ConnectionAccessContext.of(statement.getConnectContext()));
+                columns, UserIdentityAccessContext.of(user,
+                        ConnectionAccessContext.of(statement.getConnectContext())));
     }
 
     public Map<String, DataMaskSpec> evalDataMaskPolicies(DataPolicyContext policy, StatementContext statement) {
@@ -1397,16 +1399,19 @@ public static final class DataPolicyContext {
         private final AuthorizedResource.Table table;
         private final Set<String> columns;
         private final AccessContext context;
+        private final Set<String> authenticatedRoles;
 
         private DataPolicyContext(AuthorizationPlugin controller, AuthorizedSubject subject,
                 AuthorizedResource.Table table, Set<String> columns, AccessContext context) {
             this.controller = controller;
             this.subject = subject;
             this.table = table;
             this.columns = Set.copyOf(columns);
+            UserIdentity identity = AccessTranslation.userIdentityOf(subject, context);
+            this.authenticatedRoles = identity.getAuthenticatedRoles();
             Optional<String> clientIp = context.getClientIp();
             Optional<String> queryId = context.getQueryId();
-            this.context = new AccessContext() {
+            this.context = UserIdentityAccessContext.of(identity, new AccessContext() {
                 @Override
                 public Optional<String> getClientIp() {
                     return clientIp;
@@ -1416,12 +1421,13 @@ public Optional<String> getClientIp() {
                 public Optional<String> getQueryId() {
                     return queryId;
                 }
-            };
+            });
         }
 
         public boolean samePolicyScope(DataPolicyContext other) {
             return controller == other.controller && subject.equals(other.subject)
                     && table.equals(other.table) && columns.equals(other.columns)
+                    && authenticatedRoles.equals(other.authenticatedRoles)
                     && context.getClientIp().equals(other.context.getClientIp());
         }
 
@@ -1433,7 +1439,7 @@ public boolean equals(Object other) {
 
         @Override
         public int hashCode() {
-            return Objects.hash(System.identityHashCode(controller), subject, table, columns,
+            return Objects.hash(System.identityHashCode(controller), subject, table, columns, authenticatedRoles,
                     context.getClientIp(), context.getQueryId());
         }
     }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessTranslation.java

@@ -21,6 +21,7 @@
 import org.apache.doris.analysis.ResourceTypeEnum;
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.authorization.AccessAction;
+import org.apache.doris.authorization.AccessContext;
 import org.apache.doris.authorization.AccessRequirement;
 import org.apache.doris.authorization.ActionMatch;
 import org.apache.doris.authorization.AuthorizedSubject;
@@ -261,6 +262,14 @@ public static UserIdentity userIdentityOf(AuthorizedSubject subject) {
                 : UserIdentity.createAnalyzedUserIdentWithIp(subject.getUser(), subject.getHost());
     }
 
+    /** Restore session roles without changing the account-only plugin API. */
+    public static UserIdentity userIdentityOf(AuthorizedSubject subject, AccessContext context) {
+        if (context instanceof UserIdentityAccessContext) {
+            return ((UserIdentityAccessContext) context).userIdentityOf(subject);
+        }
+        return userIdentityOf(subject);
+    }
+
     /** The resource kind standing for a cloud object of {@code type}. */
     public static ResourceKind cloudKindOf(ResourceTypeEnum type) {
         switch (type) {

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Auth.java

@@ -297,36 +297,52 @@ public void checkPasswordForUserIdentity(UserIdentity userIdentity, byte[] remot
     }
 
     public Set<Role> getRolesByUserWithLdap(UserIdentity userIdentity) {
-        Set<Role> roles = Sets.newHashSet();
-        Set<String> roleNames = userRoleManager.getRolesByUser(userIdentity);
-        for (String roleName : roleNames) {
-            Role role = roleManager.getRole(roleName);
-            if (role != null) {
-                roles.add(role);
-            }
-        }
+        Set<Role> roles = getRolesByUserWithAuthentication(userIdentity);
         if (isLdapAuthEnabled()) {
             Set<Role> ldapRoles = ldapManager.getUserRoles(userIdentity.getQualifiedUser());
             if (!CollectionUtils.isEmpty(ldapRoles)) {
                 roles.addAll(ldapRoles);
             }
         }
+        return roles;
+    }
+
+    private Set<Role> getRolesByUserWithAuthentication(UserIdentity userIdentity) {
+        Set<Role> roles = Sets.newHashSet();
+        Set<String> roleNames = userRoleManager.getRolesByUser(userIdentity);
         ConnectContext ctx = ConnectContext.get();
-        if (ctx != null && userIdentity.equals(ctx.getCurrentUserIdentity())) {
-            for (String roleName : ctx.getAuthenticatedRoles()) {
-                Role role = roleManager.getRole(roleName);
-                if (role != null) {
-                    roles.add(role);
-                }
+        Set<String> sessionRoles = ctx == null
+                ? userIdentity.getAuthenticatedRoles() : ctx.getAuthenticatedRolesFor(userIdentity);
+        if (!sessionRoles.isEmpty()) {
+            roleNames = Sets.newHashSet(roleNames);
+            roleNames.addAll(sessionRoles);
+        }
+        for (String roleName : roleNames) {
+            Role role = roleManager.getRole(roleName);
+            if (role != null) {
+                roles.add(role);
             }
         }
         return roles;
     }
 
+    /** Named memberships for plugins; legacy LDAP's synthetic privilege roles stay inside built-in checks. */
+    public Set<String> getRoleNamesForAuthorization(UserIdentity user) {
+        readLock();
+        try {
+            return getRoleNamesByRoles(getRolesByUserWithAuthentication(user), false);
+        } finally {
+            readUnlock();
+        }
+    }
+
     public Set<String> getRoleNamesByUserWithLdap(UserIdentity user, boolean showUserDefaultRole) {
-        Set<Role> rolesByUserWithLdap = getRolesByUserWithLdap(user);
-        Set<String> res = Sets.newHashSetWithExpectedSize(rolesByUserWithLdap.size());
-        for (Role role : rolesByUserWithLdap) {
+        return getRoleNamesByRoles(getRolesByUserWithLdap(user), showUserDefaultRole);
+    }
+
+    private Set<String> getRoleNamesByRoles(Set<Role> roles, boolean showUserDefaultRole) {
+        Set<String> res = Sets.newHashSetWithExpectedSize(roles.size());
+        for (Role role : roles) {
             String roleName = role.getRoleName();
             if (showUserDefaultRole || !roleName.startsWith(RoleManager.DEFAULT_ROLE_PREFIX)) {
                 res.add(roleName);

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/ConnectionAccessContext.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.mysql.privilege;
 
+import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.authorization.AccessContext;
 import org.apache.doris.common.util.DebugUtil;
 import org.apache.doris.qe.ConnectContext;
@@ -62,6 +63,14 @@ static AccessContext of(ConnectContext connection) {
         return connection == null ? AccessContext.NONE : new ConnectionAccessContext(connection);
     }
 
+    UserIdentity userIdentityWithRoles(UserIdentity identity) {
+        if (!identity.equals(connection.getCurrentUserIdentity()) || connection.getAuthenticatedRoles().isEmpty()) {
+            return identity;
+        }
+        return UserIdentity.createAnalyzedUserIdentWithRoles(identity.getUser(), identity.getHost(),
+                identity.isDomain(), connection.getAuthenticatedRolesFor(identity));
+    }
+
     @Override
     public Optional<String> getClientIp() {
         return Optional.ofNullable(Strings.emptyToNull(connection.getRemoteIP()));

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/EngineAuthorizationContext.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.mysql.privilege;
 
+import org.apache.doris.authorization.AccessContext;
 import org.apache.doris.authorization.AccessRequirement;
 import org.apache.doris.authorization.AuthorizedResource;
 import org.apache.doris.authorization.AuthorizedSubject;
@@ -74,7 +75,8 @@ public void servedBy(AuthorizationPlugin source) {
      */
     @Override
     public Set<String> rolesOf(AuthorizedSubject subject) {
-        return auth.getRolesByUser(AccessTranslation.userIdentityOf(subject), false);
+        return auth.getRoleNamesForAuthorization(
+                AccessTranslation.userIdentityOf(subject, manager.contextOfCheckInFlight()));
     }
 
     @Override
@@ -90,7 +92,8 @@ public boolean grantedByGlobalScopeAuthority(AuthorizedSubject subject, AccessRe
         // on the thread: the source that governs instance scope may decide from the client address, and this
         // question is about the same statement. A check can reach here before its connection is installed on
         // the thread - the HTTP cookie path does - where the thread holds another request's, or none.
-        return manager.decide(AccessTranslation.userIdentityOf(subject), AuthorizedResource.global(),
-                requirement, manager.contextOfCheckInFlight());
+        AccessContext context = manager.contextOfCheckInFlight();
+        return manager.decide(AccessTranslation.userIdentityOf(subject, context), AuthorizedResource.global(),
+                requirement, context);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/InternalAuthorizationPlugin.java

@@ -73,7 +73,7 @@ public String name() {
     @Override
     public void checkPrivilege(AuthorizedSubject subject, AuthorizedResource resource,
             AccessRequirement requirement, AccessContext context) throws AccessDeniedException {
-        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject);
+        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject, context);
         PrivPredicate wanted = AccessTranslation.privPredicateOf(requirement);
         switch (resource.getKind()) {
             case GLOBAL:
@@ -163,7 +163,7 @@ public Map<String, DataMaskSpec> getDataMasks(AuthorizedSubject subject, Authori
     public List<RowFilterSpec> getRowFilters(AuthorizedSubject subject, AuthorizedResource.Table table,
             AccessContext context) {
         List<RowPolicy> policies = Env.getCurrentEnv().getPolicyMgr().getUserPolicies(table.getCatalog(),
-                table.getDatabase(), table.getTable(), AccessTranslation.userIdentityOf(subject));
+                table.getDatabase(), table.getTable(), AccessTranslation.userIdentityOf(subject, context));
         ImmutableList.Builder<RowFilterSpec> specs = ImmutableList.builderWithExpectedSize(policies.size());
         for (RowPolicy policy : policies) {
             try {

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/LegacyAccessControllerPlugin.java

@@ -95,7 +95,7 @@ public String name() {
     @Override
     public void checkPrivilege(AuthorizedSubject subject, AuthorizedResource resource,
             AccessRequirement requirement, AccessContext context) throws AccessDeniedException {
-        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject);
+        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject, context);
         PrivPredicate wanted = AccessTranslation.privPredicateOf(requirement);
         switch (resource.getKind()) {
             case GLOBAL:
@@ -191,7 +191,7 @@ private void refuseUnless(boolean allowed, AuthorizedSubject subject, Authorized
     public List<RowFilterSpec> getRowFilters(AuthorizedSubject subject, AuthorizedResource.Table table,
             AccessContext context) {
         List<RowFilterSpec> filters = controller.evalRowFilterPolicies(
-                AccessTranslation.userIdentityOf(subject), table.getCatalog(), table.getDatabase(),
+                AccessTranslation.userIdentityOf(subject, context), table.getCatalog(), table.getDatabase(),
                 table.getTable());
         // Through a wildcard so that reading an element compiles to no cast of our own: the checked crossing
         // below has to be what reports a wrong type, not a checkcast javac inserted a frame earlier.
@@ -211,7 +211,7 @@ public List<RowFilterSpec> getRowFilters(AuthorizedSubject subject, AuthorizedRe
     @Override
     public Map<String, DataMaskSpec> getDataMasks(AuthorizedSubject subject, AuthorizedResource.Table table,
             Set<String> columns, AccessContext context) {
-        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject);
+        UserIdentity currentUser = AccessTranslation.userIdentityOf(subject, context);
         Map<String, DataMaskSpec> masks = new HashMap<>();
         for (String column : columns) {
             // One question per column, which is what the older interface offers. A source reached over the