[fix](auth) Fix CTE privilege bypass due to shared privChecked flag on StatementContext

fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java

@@ -120,6 +120,8 @@ public class CascadesContext implements ScheduleContext {
 
     private boolean isLeadingDisableJoinReorder = false;
 
+    private boolean privChecked = false;
+
     private final Map<String, Hint> hintMap = Maps.newLinkedHashMap();
     private final ThreadLocal<Boolean> showPlanProcess = new ThreadLocal<>();
 
@@ -520,6 +522,14 @@ public void setLeadingDisableJoinReorder(boolean leadingDisableJoinReorder) {
         isLeadingDisableJoinReorder = leadingDisableJoinReorder;
     }
 
+    public boolean isPrivChecked() {
+        return privChecked;
+    }
+
+    public void setPrivChecked(boolean privChecked) {
+        this.privChecked = privChecked;
+    }
+
     public Map<String, Hint> getHintMap() {
         return hintMap;
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -266,8 +266,6 @@ public enum TableFrom {
 
     private final Map<MvccTableInfo, MvccSnapshot> snapshots = Maps.newHashMap();
 
-    private boolean privChecked;
-
     // if greater than 0 means the duration has used
     private long materializedViewRewriteDuration = 0L;
 
@@ -1029,14 +1027,6 @@ public void setGroupCommitMergeBackend(
         this.groupCommitMergeBackend = groupCommitMergeBackend;
     }
 
-    public boolean isPrivChecked() {
-        return privChecked;
-    }
-
-    public void setPrivChecked(boolean privChecked) {
-        this.privChecked = privChecked;
-    }
-
     public List<Backend> getUsedBackendsDistributing() {
         return usedBackendsDistributing;
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CheckPrivileges.java

@@ -46,20 +46,24 @@
 
 /**
  * CheckPrivileges
- * This rule should only check once, because after check would set setPrivChecked in statementContext
+ * This rule should only check once per CascadesContext, because after check would set privChecked
+ * on the CascadesContext. Using CascadesContext-level (instead of StatementContext-level) flag
+ * ensures CTE producer and consumer subtrees are checked independently, while still preventing
+ * duplicate checks within the same subtree (e.g., after view inlining).
  */
 public class CheckPrivileges extends ColumnPruning {
     private JobContext jobContext;
 
     @Override
     public Plan rewriteRoot(Plan plan, JobContext jobContext) {
-        // Only enter once, if repeated, the permissions of the table in the view will be checked
-        if (jobContext.getCascadesContext().getStatementContext().isPrivChecked()) {
+        // Only enter once per CascadesContext. If repeated within the same context,
+        // the permissions of the table in the view will be checked incorrectly.
+        if (jobContext.getCascadesContext().isPrivChecked()) {
             return plan;
         }
         this.jobContext = jobContext;
         super.rewriteRoot(plan, jobContext);
-        jobContext.getCascadesContext().getStatementContext().setPrivChecked(true);
+        jobContext.getCascadesContext().setPrivChecked(true);
         // don't rewrite plan, because Reorder expect no LogicalProject on LogicalJoin
         return plan;
     }

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

@@ -207,8 +207,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
             }
         } catch (Exception e) {
             try {
-                new DeleteFromUsingCommand(nameParts, tableAlias, isTempPart, partitions,
-                        logicalQuery, Optional.empty(), false).run(ctx, executor);
+                // delete not need select priv, skip auth for the fallback planner
+                ctx.setSkipAuth(true);
+                try {
+                    new DeleteFromUsingCommand(nameParts, tableAlias, isTempPart, partitions,
+                            logicalQuery, Optional.empty(), false).run(ctx, executor);
+                } finally {
+                    ctx.setSkipAuth(originalIsSkipAuth);
+                }
                 return;
             } catch (Exception e2) {
                 LOG.warn("delete from command failed", e2);
@@ -219,8 +225,14 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
         // if table's enable_mow_light_delete is false, use `DeleteFromUsingCommand`
         if (olapTable.getKeysType() == KeysType.UNIQUE_KEYS && olapTable.getEnableUniqueKeyMergeOnWrite()
                 && !olapTable.getEnableMowLightDelete()) {
-            new DeleteFromUsingCommand(nameParts, tableAlias, isTempPart, partitions, logicalQuery,
-                    Optional.empty(), false).run(ctx, executor);
+            // delete not need select priv, skip auth for the fallback planner
+            ctx.setSkipAuth(true);
+            try {
+                new DeleteFromUsingCommand(nameParts, tableAlias, isTempPart, partitions, logicalQuery,
+                        Optional.empty(), false).run(ctx, executor);
+            } finally {
+                ctx.setSkipAuth(originalIsSkipAuth);
+            }
             return;
         }
 

fe/fe-core/src/test/java/org/apache/doris/nereids/privileges/TestCheckPrivileges.java

@@ -223,6 +223,77 @@ public void testPrivilegesAndPolicies() throws Exception {
         });
     }
 
+    @Test
+    public void testCtePrivilegeCheck() throws Exception {
+        FeConstants.runningUnitTest = true;
+        // Use the same external catalog for privilege control infrastructure
+        String catalogProvider
+                = "org.apache.doris.nereids.privileges.TestCheckPrivileges$CustomCatalogProvider";
+        String accessControllerFactory
+                = "org.apache.doris.nereids.privileges.CustomAccessControllerFactory";
+        String catalog = "custom_catalog_cte";
+
+        createCatalog("create catalog " + catalog + " properties("
+                + " \"type\"=\"test\","
+                + " \"catalog_provider.class\"=\"" + catalogProvider + "\","
+                + " \"" + CatalogMgr.ACCESS_CONTROLLER_CLASS_PROP + "\"=\"" + accessControllerFactory + "\""
+                + ")");
+
+        // Create internal OLAP tables (LogicalOlapScan supports withRelationId needed by CTE)
+        String cteDb = "cte_priv_db";
+        createDatabase(cteDb);
+        useDatabase(cteDb);
+        createTable("create table allowed_tbl (id int, name varchar(50)) "
+                + "distributed by hash(id) buckets 1 "
+                + "properties('replication_num' = '1')");
+        createTable("create table denied_tbl (id int, name varchar(50)) "
+                + "distributed by hash(id) buckets 1 "
+                + "properties('replication_num' = '1')");
+
+        String user = "test_cte_privilege_user";
+        addUser(user, true);
+        useUser(user);
+
+        try {
+            List<MakeTablePrivileges> privileges = ImmutableList.of(
+                    MakePrivileges.table("internal", cteDb, "allowed_tbl").allowSelectTable(user)
+            );
+
+            AccessControllerManager accessManager = Env.getCurrentEnv().getAccessManager();
+            CatalogAccessController catalogAccessController = accessManager.getAccessControllerOrDefault(catalog);
+            AccessControllerManager spyAccessManager = Mockito.spy(accessManager);
+            Mockito.doReturn(catalogAccessController).when(spyAccessManager)
+                    .getAccessControllerOrDefault("internal");
+            Deencapsulation.setField(Env.getCurrentEnv(), "accessManager", spyAccessManager);
+
+            withPrivileges(privileges, () -> {
+                // CTE with authorized table should succeed
+                query("with cte as (select * from allowed_tbl) select * from cte");
+
+                // CTE with unauthorized table should fail (this was the bug:
+                // consumer was checked first, setting privChecked=true on shared StatementContext,
+                // so producer's CheckPrivileges was skipped)
+                Assertions.assertThrows(AnalysisException.class, () ->
+                        query("with cte as (select * from denied_tbl) select * from cte")
+                );
+
+                // CTE referenced twice forces no-inline path through RewriteCteChildren
+                Assertions.assertThrows(AnalysisException.class, () ->
+                        query("with cte as (select * from denied_tbl) "
+                                + "select * from cte union all select * from cte")
+                );
+
+                // CTE authorized + direct unauthorized table should fail
+                Assertions.assertThrows(AnalysisException.class, () ->
+                        query("with cte as (select * from allowed_tbl) "
+                                + "select * from cte union all select * from denied_tbl")
+                );
+            });
+        } finally {
+            useUser("root");
+        }
+    }
+
     private void query(String sql) {
         PlanChecker.from(connectContext)
                 .parse(sql)

fe/fe-core/src/test/java/org/apache/doris/nereids/privileges/TestCtePrivCheckGranularity.java

@@ -0,0 +1,97 @@
+// 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.doris.nereids.privileges;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+/**
+ * Tests that privChecked flag is per-CascadesContext, not per-StatementContext.
+ * This ensures CTE producer and consumer subtrees check privileges independently.
+ */
+public class TestCtePrivCheckGranularity {
+
+    @Test
+    public void testPrivCheckedDefaultFalse() {
+        CascadesContext ctx = CascadesContext.initTempContext();
+        Assertions.assertFalse(ctx.isPrivChecked(), "privChecked should default to false");
+    }
+
+    @Test
+    public void testPrivCheckedSetAndGet() {
+        CascadesContext ctx = CascadesContext.initTempContext();
+
+        ctx.setPrivChecked(true);
+        Assertions.assertTrue(ctx.isPrivChecked());
+
+        ctx.setPrivChecked(false);
+        Assertions.assertFalse(ctx.isPrivChecked());
+    }
+
+    @Test
+    public void testSubtreeContextHasIndependentPrivChecked() {
+        CascadesContext parentCtx = CascadesContext.initTempContext();
+
+        // Simulate CTE: create subtree context (shares StatementContext but different CascadesContext)
+        CascadesContext subtreeCtx = CascadesContext.newSubtreeContext(
+                Optional.empty(), parentCtx, parentCtx.getRewritePlan(), PhysicalProperties.ANY);
+
+        // Both should share the same StatementContext
+        Assertions.assertSame(parentCtx.getStatementContext(), subtreeCtx.getStatementContext(),
+                "Parent and subtree should share the same StatementContext");
+
+        // Both should start with privChecked = false
+        Assertions.assertFalse(parentCtx.isPrivChecked());
+        Assertions.assertFalse(subtreeCtx.isPrivChecked());
+
+        // Setting privChecked on subtree should NOT affect parent
+        subtreeCtx.setPrivChecked(true);
+        Assertions.assertTrue(subtreeCtx.isPrivChecked(), "subtree privChecked should be true");
+        Assertions.assertFalse(parentCtx.isPrivChecked(), "parent privChecked should still be false");
+
+        // Setting privChecked on parent should NOT affect subtree
+        parentCtx.setPrivChecked(true);
+        Assertions.assertTrue(parentCtx.isPrivChecked());
+        Assertions.assertTrue(subtreeCtx.isPrivChecked()); // still true from before
+    }
+
+    @Test
+    public void testTwoSubtreeContextsAreIndependent() {
+        CascadesContext parentCtx = CascadesContext.initTempContext();
+
+        // Simulate CTE consumer and producer: two subtree contexts from same parent
+        CascadesContext consumerCtx = CascadesContext.newSubtreeContext(
+                Optional.empty(), parentCtx, parentCtx.getRewritePlan(), PhysicalProperties.ANY);
+        CascadesContext producerCtx = CascadesContext.newSubtreeContext(
+                Optional.empty(), parentCtx, parentCtx.getRewritePlan(), PhysicalProperties.ANY);
+
+        // Consumer runs CheckPrivileges first, sets its flag
+        consumerCtx.setPrivChecked(true);
+
+        // Producer should NOT see consumer's flag — this is the bug fix
+        Assertions.assertFalse(producerCtx.isPrivChecked(),
+                "Producer CascadesContext should have independent privChecked from consumer");
+        Assertions.assertFalse(parentCtx.isPrivChecked(),
+                "Parent CascadesContext should not be affected by consumer's privChecked");
+    }
+}