[fix](auth) Add missing privilege checks for several Nereids commands

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AddConstraintCommand.java

@@ -24,9 +24,12 @@
 import org.apache.doris.catalog.constraint.PrimaryKeyConstraint;
 import org.apache.doris.catalog.constraint.UniqueConstraint;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Pair;
 import org.apache.doris.info.TableNameInfoUtils;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.properties.PhysicalProperties;
@@ -71,6 +74,7 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         TableNameInfo tableNameInfo = TableNameInfoUtils.fromCatalogDb(
                 table.getDatabase().getCatalog(), table.getDatabase(), table);
         ImmutableList<String> columns = columnsAndTable.first;
+        checkAlterPriv(ctx, tableNameInfo);
 
         Pair<ImmutableList<String>, TableNameInfo> referencedColumnsAndTable = null;
         if (constraint.isForeignKey()) {
@@ -79,6 +83,8 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
             TableIf refTable = refColumnsAndTable.second;
             TableNameInfo refTableInfo = TableNameInfoUtils.fromCatalogDb(
                     refTable.getDatabase().getCatalog(), refTable.getDatabase(), refTable);
+            // a foreign key also registers a reverse reference on the referenced table
+            checkAlterPriv(ctx, refTableInfo);
             referencedColumnsAndTable = Pair.of(refColumnsAndTable.first, refTableInfo);
         }
         if (constraint.isForeignKey()) {
@@ -97,6 +103,16 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
     }
 
+    private void checkAlterPriv(ConnectContext ctx, TableNameInfo tableNameInfo)
+            throws org.apache.doris.common.AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, tableNameInfo.getCtl(),
+                tableNameInfo.getDb(), tableNameInfo.getTbl(), PrivPredicate.ALTER)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "ALTER",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(),
+                    tableNameInfo.getDb() + ": " + tableNameInfo.getTbl());
+        }
+    }
+
     private void addConstraintAndInvalidate(
             TableNameInfo tableNameInfo, org.apache.doris.catalog.constraint.Constraint constraint)
             throws Exception {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminRotateTdeRootKeyCommand.java

@@ -17,8 +17,12 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.encryption.KeyManagerInterface;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -52,6 +56,7 @@ public AdminRotateTdeRootKeyCommand(Map<String, String> properties) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         KeyManagerInterface keyManager = ctx.getEnv().getKeyManager();
         if (keyManager != null) {
             keyManager.rotateRootKey(properties);
@@ -60,6 +65,16 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         }
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitAdminRotateTdeRootKeyCommand(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AdminSetEncryptionRootKeyCommand.java

@@ -17,9 +17,13 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.encryption.EncryptionKey;
 import org.apache.doris.encryption.RootKeyInfo;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -63,6 +67,11 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
      * validate
      */
     public void validate() throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
         if (properties == null || properties.isEmpty()) {
             throw new AnalysisException("The properties must not be empty");
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CancelWarmUpJobCommand.java

@@ -17,9 +17,13 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.analyzer.UnboundSlot;
 import org.apache.doris.nereids.trees.expressions.EqualTo;
 import org.apache.doris.nereids.trees.expressions.Expression;
@@ -59,6 +63,11 @@ public long getJobId() {
      * @throws AnalysisException check whether this sql is legal
      */
     public void validate(ConnectContext ctx) throws AnalysisException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
         if (!Config.isCloudMode()) {
             throw new AnalysisException("The sql is illegal in disk mode ");
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateDictionaryCommand.java

@@ -18,7 +18,12 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.dictionary.LayoutType;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateDictionaryInfo;
@@ -60,12 +65,20 @@ public StmtType stmtType() {
     }
 
     @Override
-    public void run(ConnectContext ctx, StmtExecutor executor) {
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         try {
             // 1. Validate the dictionary info. names and existence.
             createDictionaryInfo.validateAndSet(ctx);
+        } catch (Exception e) {
+            LOG.warn("Failed to create dictionary: {}", e.getMessage());
+            throw new AnalysisException("Failed to create dictionary: " + e.getMessage());
+        }
+
+        // 2. Check auth. Must run after validateAndSet(), which fills in the default catalog/db names.
+        checkAuth(ctx);
 
-            // 2. Create dictionary and save it in manager. it will schedule data load.
+        try {
+            // 3. Create dictionary and save it in manager. it will schedule data load.
             ctx.getEnv().getDictionaryManager().createDictionary(ctx, createDictionaryInfo);
 
             LOG.info("Created dictionary {} in {} from {}", createDictionaryInfo.getDictName(),
@@ -75,4 +88,25 @@ public void run(ConnectContext ctx, StmtExecutor executor) {
             throw new AnalysisException("Failed to create dictionary: " + e.getMessage());
         }
     }
+
+    /**
+     * A dictionary is created in the internal catalog and its data is loaded by an internal task,
+     * so require CREATE on the dictionary itself and SELECT on the source table. Without the latter
+     * the dictionary would expose data the creator can not read.
+     */
+    private void checkAuth(ConnectContext ctx) throws org.apache.doris.common.AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                createDictionaryInfo.getDbName(), createDictionaryInfo.getDictName(), PrivPredicate.CREATE)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "CREATE");
+        }
+
+        String srcCtl = createDictionaryInfo.getSourceCtlName();
+        String srcDb = createDictionaryInfo.getSourceDbName();
+        String srcTbl = createDictionaryInfo.getSourceTableName();
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, srcCtl, srcDb, srcTbl,
+                PrivPredicate.SELECT)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SELECT",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), srcDb + ": " + srcTbl);
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropCatalogRecycleBinCommand.java

@@ -18,6 +18,9 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -71,9 +74,21 @@ public DropCatalogRecycleBinCommand(IdType idType, long id) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         Env.getCurrentEnv().dropCatalogRecycleBin(idType, id);
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws org.apache.doris.common.AnalysisException {
+        // Erasing from the recycle bin is irreversible and takes a raw object id, so it can not be
+        // authorized at db/table level. Restrict it to ADMIN, same as the other catalog-wide admin ops.
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitDropCatalogRecycleBinCommand(this, context);

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropConstraintCommand.java

@@ -22,8 +22,11 @@
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.constraint.Constraint;
 import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.info.TableNameInfoUtils;
 import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.NereidsPlanner;
 import org.apache.doris.nereids.analyzer.UnboundRelation;
 import org.apache.doris.nereids.exceptions.AnalysisException;
@@ -75,6 +78,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
                     + "falling back to name-based lookup: {}", name, e.getMessage());
             tableNameInfo = extractTableNameFromPlan(ctx);
         }
+        // must be checked on both paths above: table resolution failing (which includes an
+        // authorization failure) falls back to a name-only lookup that binds nothing.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, tableNameInfo.getCtl(),
+                tableNameInfo.getDb(), tableNameInfo.getTbl(), PrivPredicate.ALTER)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "ALTER",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(),
+                    tableNameInfo.getDb() + ": " + tableNameInfo.getTbl());
+        }
         Constraint constraint = Env.getCurrentEnv().getConstraintManager().getConstraint(tableNameInfo, name);
         if (constraint == null) {
             throw new AnalysisException(

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropDictionaryCommand.java

@@ -18,6 +18,11 @@
 package org.apache.doris.nereids.trees.plans.commands;
 
 import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -57,10 +62,15 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
     }
 
     @Override
-    public void run(ConnectContext ctx, StmtExecutor executor) {
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         if (dbName == null) { // use current database
             dbName = ctx.getDatabase();
         }
+        // check auth. dictionaries always live in the internal catalog.
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                dbName, dictName, PrivPredicate.DROP)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "DROP");
+        }
         try {
             ctx.getEnv().getDictionaryManager().dropDictionary(ctx, dbName, dictName, ifExists);
         } catch (Exception e) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropStageCommand.java

@@ -20,6 +20,10 @@
 import org.apache.doris.analysis.StmtType;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.catalog.CloudEnv;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
@@ -45,9 +49,20 @@ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
 
     @Override
     public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
         ((CloudEnv) Env.getCurrentEnv()).dropStage(this);
     }
 
+    /**
+     * validate
+     */
+    public void validate() throws UserException {
+        // check auth. keep it aligned with CreateStageCommand.
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+    }
+
     @Override
     public StmtType stmtType() {
         return StmtType.DROP;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.nereids.trees.plans.commands;
 
+import org.apache.doris.analysis.ResourceTypeEnum;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
@@ -33,6 +34,7 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Triple;
 import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.WarmUpItem;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@@ -139,6 +141,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         handleWarmUp(ctx, executor);
     }
 
+    private void checkComputeGroupUsage(ConnectContext ctx, String computeGroup) throws AnalysisException {
+        if (!Env.getCurrentEnv().getAccessManager().checkCloudPriv(ctx.getCurrentUserIdentity(),
+                computeGroup, PrivPredicate.USAGE, ResourceTypeEnum.CLUSTER)) {
+            throw new AnalysisException("USAGE denied to user '" + ctx.getQualifiedUser() + "'@'"
+                    + ctx.getRemoteIP() + "' for compute group '" + computeGroup + "'");
+        }
+    }
+
     private void checkWarmupCgs(CloudSystemInfoService cloudSys) throws AnalysisException {
         if (!Strings.isNullOrEmpty(srcCluster)) {
             CloudComputeGroupMeta srcCg = cloudSys.getComputeGroupByName(srcCluster);
@@ -180,6 +190,13 @@ public void validate(ConnectContext connectContext) throws UserException {
             throw new UserException("The sql is just support in cloud mode");
         }
 
+        // check auth. warming up moves data between compute groups, so require USAGE on both ends
+        // instead of global ADMIN. Keep it aligned with UseCloudClusterCommand.
+        checkComputeGroupUsage(connectContext, dstCluster);
+        if (!Strings.isNullOrEmpty(srcCluster)) {
+            checkComputeGroupUsage(connectContext, srcCluster);
+        }
+
         CloudSystemInfoService cloudSys = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
         if (!cloudSys.containClusterName(dstCluster)) {
             throw new AnalysisException("The dstClusterName " + dstCluster + " doesn't exist");
@@ -226,6 +243,12 @@ public void validate(ConnectContext connectContext) throws UserException {
                 if (table == null) {
                     ErrorReport.reportAnalysisException(ErrorCode.ERR_BAD_TABLE_ERROR, tableNameInfo.getTbl());
                 }
+                if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(connectContext, tableNameInfo.getCtl(),
+                        dbName, tableNameInfo.getTbl(), PrivPredicate.SELECT)) {
+                    ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SELECT",
+                            connectContext.getQualifiedUser(), connectContext.getRemoteIP(),
+                            dbName + ": " + tableNameInfo.getTbl());
+                }
                 if (partitionName.length() != 0 && !table.containsPartition(partitionName)) {
                     throw new AnalysisException("The partition " + partitionName + " doesn't exist");
                 }

fe/fe-core/src/test/java/org/apache/doris/cloud/WarmUpClusterOnTablesParseTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.cloud;
 
+import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.OnTablesFilter.TableFilterRule;
 import org.apache.doris.cloud.OnTablesFilter.TableFilterRule.RuleType;
@@ -66,6 +67,11 @@ public static void init() throws Exception {
         originalSystemInfo = getField(env, Env.class, "systemInfo");
         connectContext = new ConnectContext();
         connectContext.setEnv(env);
+        // this test covers ON TABLES parsing and validation, not authorization, so give the
+        // context an identity and let it bypass the privilege checks in validate()
+        connectContext.setCurrentUserIdentity(UserIdentity.ROOT);
+        connectContext.setNoAuth(true);
+        connectContext.setSkipAuth(true);
         connectContext.setThreadLocalInfo();
     }