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

### What problem does this PR solve?

Issue Number: N/A

Related PR: #44621

Problem Summary:

When a query uses a Common Table Expression (CTE) that references an unauthorized table,
the privilege check is incorrectly skipped, allowing the query to succeed without proper
authorization.

**Root cause:** `CheckPrivileges.rewriteRoot()` uses `StatementContext.isPrivChecked()` as
a guard to run only once. However, CTE processing (`RewriteCteChildren.visitLogicalCTEAnchor()`)
creates separate `CascadesContext` subtrees for consumer and producer that **share** the same
`StatementContext`. The consumer subtree is processed first, its `CheckPrivileges` sets
`privChecked=true` on the shared `StatementContext`, and when the producer subtree runs,
`CheckPrivileges` sees the flag and skips — leaving unauthorized tables in the CTE body
unchecked.

```
StatementContext (shared)
├─ CascadesContext (consumer) ── CheckPrivileges runs ──> sets privChecked=true
└─ CascadesContext (producer) ── CheckPrivileges sees true ──> SKIPPED (BUG)
```

**Fix:** Move the `privChecked` flag from `StatementContext` to `CascadesContext`.
Since `newSubtreeContext()` creates a **new** `CascadesContext` instance for each CTE
subtree, each subtree now has its own independent flag. This ensures both producer and
consumer run their own privilege checks. Within the same `CascadesContext`, the flag
still prevents duplicate checks (e.g., after view inlining — preserving #44621 semantics).

```
StatementContext (shared, no privChecked)
├─ CascadesContext (consumer) ── privChecked=false ── CheckPrivileges runs ── privChecked=true
└─ CascadesContext (producer) ── privChecked=false ── CheckPrivileges runs ── privChecked=true ✓
```

| File | Change Description |
|------|-------------------|
| `CascadesContext.java` | Add `privChecked` boolean field + getter/setter |
| `StatementContext.java` | Remove `privChecked` field and accessors |
| `CheckPrivileges.java` | Read/write `privChecked` on `CascadesContext` instead of `StatementContext` |
| `TestCheckPrivileges.java` | Add `testCtePrivilegeCheck()` with 4 CTE authorization scenarios |
| `TestCtePrivCheckGranularity.java` (new) | 4 unit tests verifying per-CascadesContext flag independence |
| `test_cte_privilege_check.groovy` (new) | 6 regression test cases covering CTE privilege bypass |

### Release note

Fixed a privilege bypass where queries using CTE (Common Table Expression) could access
unauthorized tables without being denied. The privilege check was incorrectly shared across
CTE subtrees, causing the producer's check to be skipped after the consumer's check ran.

### Check List (For Author)

- Test
    - [x] Regression test
    - [x] Unit Test

- Behavior changed:
    - [x] No.

- Does this need documentation?
    - [x] No.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: seawinde

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

@@ -263,8 +263,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;
 
@@ -1014,14 +1012,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/test/java/org/apache/doris/nereids/privileges/TestCheckPrivileges.java

@@ -225,6 +225,76 @@ 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);
+
+        List<MakeTablePrivileges> privileges = ImmutableList.of(
+                MakePrivileges.table("internal", cteDb, "allowed_tbl").allowSelectTable(user)
+        );
+
+        AccessControllerManager accessManager = Env.getCurrentEnv().getAccessManager();
+        CatalogAccessController catalogAccessController = accessManager.getAccessControllerOrDefault(catalog);
+        new Expectations(accessManager) {
+            {
+                accessManager.getAccessControllerOrDefault("internal");
+                minTimes = 0;
+                result = catalogAccessController;
+            }
+        };
+
+        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")
+            );
+        });
+    }
+
     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");
+    }
+}