[fix](auth)support check priv when tvf use resource

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

@@ -53,6 +53,7 @@ public void analyze(Analyzer analyzer) throws UserException {
         super.analyze(analyzer);
 
         // check auth
+        // check if can alter policy and use storage_resource
         if (!Env.getCurrentEnv().getAccessManager()
                 .checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,

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

@@ -102,6 +102,7 @@ public void analyze(Analyzer analyzer) throws UserException {
                             + "Enable it by setting 'enable_storage_policy=true' in fe.conf");
                 }
                 // check auth
+                // check if can create policy and use storage_resource
                 if (!Env.getCurrentEnv().getAccessManager()
                         .checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
                     ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR,

fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java

@@ -83,6 +83,10 @@ public enum ErrorCode {
 
     ERR_SPECIFIC_ALL_ACCESS_DENIED_ERROR(1223, new byte[] {'4', '2', '0', '0', '0'}, "Access denied; you need all "
             + " %s privilege(s) for this operation"),
+
+    ERR_RESOURCE_ACCESS_DENIED_ERROR(1222, new byte[]{'4', '2', '0', '0', '0'}, "Access denied; you need (at least "
+            + "one of) the (%s) privilege(s) on resource %s for this operation"),
+
     ERR_LOCAL_VARIABLE(1228, new byte[]{'H', 'Y', '0', '0', '0'}, "Variable '%s' is a SESSION variable and can't be "
             + "used with SET GLOBAL"),
     ERR_GLOBAL_VARIABLE(1229, new byte[]{'H', 'Y', '0', '0', '0'}, "Variable '%s' is a GLOBAL variable and should be "

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

@@ -27,9 +27,11 @@
 import org.apache.doris.nereids.rules.analysis.UserAuthentication;
 import org.apache.doris.nereids.trees.expressions.Slot;
 import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
 import org.apache.doris.nereids.trees.plans.logical.LogicalView;
 import org.apache.doris.qe.ConnectContext;
 
@@ -62,6 +64,13 @@ public Plan visitLogicalView(LogicalView<? extends Plan> view, PruneContext cont
         return view;
     }
 
+    @Override
+    public Plan visitLogicalTVFRelation(LogicalTVFRelation tvfRelation, PruneContext context) {
+        TableValuedFunction tvf = tvfRelation.getFunction();
+        tvf.checkAuth(jobContext.getCascadesContext().getConnectContext());
+        return super.visitLogicalTVFRelation(tvfRelation, context);
+    }
+
     @Override
     public Plan visitLogicalRelation(LogicalRelation relation, PruneContext context) {
         if (relation instanceof LogicalCatalogRelation) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/TableValuedFunction.java

@@ -31,6 +31,7 @@
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
 import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
 import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.Statistics;
@@ -103,6 +104,10 @@ public final FunctionGenTable getTable() {
         return tableCache.get();
     }
 
+    public final void checkAuth(ConnectContext ctx) {
+        getCatalogFunction().checkAuth(ctx);
+    }
+
     @Override
     public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
         return visitor.visitTableValuedFunction(this, context);

fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java

@@ -33,6 +33,7 @@
 import org.apache.doris.catalog.Table;
 import org.apache.doris.catalog.Type;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.BrokerUtil;
@@ -41,6 +42,7 @@
 import org.apache.doris.common.util.NetUtils;
 import org.apache.doris.common.util.Util;
 import org.apache.doris.datasource.tvf.source.TVFScanNode;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanNode;
 import org.apache.doris.proto.InternalService;
@@ -125,6 +127,9 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
     protected String filePath;
 
     protected TFileFormatType fileFormatType;
+
+    protected Optional<String> resourceName = Optional.empty();
+
     private TFileCompressType compressionType;
     private String headerType = "";
 
@@ -181,6 +186,7 @@ protected Map<String, String> parseCommonProperties(Map<String, String> properti
             if (resource == null) {
                 throw new AnalysisException("Can not find resource: " + properties.get("resource"));
             }
+            this.resourceName = Optional.of(properties.get("resource"));
             mergedProperties = resource.getCopiedProperties();
         }
         mergedProperties.putAll(properties);
@@ -564,5 +570,16 @@ private boolean isFileContentEmpty(TBrokerFileStatus fileStatus) {
         }
         return false;
     }
+
+    public void checkAuth(ConnectContext ctx) {
+        if (resourceName.isPresent()) {
+            if (!Env.getCurrentEnv().getAccessManager()
+                    .checkResourcePriv(ctx, resourceName.get(), PrivPredicate.USAGE)) {
+                String message = ErrorCode.ERR_RESOURCE_ACCESS_DENIED_ERROR.formatErrorMsg(
+                        PrivPredicate.USAGE.getPrivs().toString(), resourceName.get());
+                throw new org.apache.doris.nereids.exceptions.AnalysisException(message);
+            }
+        }
+    }
 }
 

fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java

@@ -24,6 +24,7 @@
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.planner.PlanNodeId;
 import org.apache.doris.planner.ScanNode;
+import org.apache.doris.qe.ConnectContext;
 
 import java.util.List;
 import java.util.Map;
@@ -86,4 +87,8 @@ public static TableValuedFunctionIf getTableFunction(String funcName, Map<String
     public abstract List<Column> getTableColumns() throws AnalysisException;
 
     public abstract ScanNode getScanNode(PlanNodeId id, TupleDescriptor desc);
+
+    public void checkAuth(ConnectContext ctx) {
+
+    }
 }

regression-test/suites/external_table_p0/tvf/test_s3_tvf_with_resource.groovy

@@ -176,4 +176,52 @@ suite("test_s3_tvf_with_resource", "p0") {
     } finally {
     }
 
+    // test auth
+    String user = 'test_s3_tvf_with_resource_user'
+    String pwd = 'C123_567p'
+    String viewName = "test_s3_tvf_with_resource_view"
+    try_sql("DROP USER ${user}")
+    sql """CREATE USER '${user}' IDENTIFIED BY '${pwd}'"""
+    String dbName = context.config.getDbNameByFile(context.file)
+    sql """grant select_priv on ${dbName}.${viewName} to ${user}"""
+    sql "drop view if exists ${viewName}"
+    sql """
+        create view ${viewName} as
+        SELECT * FROM S3 (
+                           "uri" = "https://${bucket}.${s3_endpoint}/regression/tvf/test_hive_text.text",
+                           "format" = "hive_text",
+                           "csv_schema"="k1:int;k2:string;k3:double",
+                           "resource" = "${resource_name}"
+                       )  where k1 > 100  order by k3,k2,k1;
+        """
+
+    //cloud-mode
+    if (isCloudMode()) {
+        def clusters = sql " SHOW CLUSTERS; "
+        assertTrue(!clusters.isEmpty())
+        def validCluster = clusters[0][0]
+        sql """GRANT USAGE_PRIV ON CLUSTER ${validCluster} TO ${user}""";
+    }
+    // not have usage priv, can not select tvf with resource
+    connect(user=user, password="${pwd}", url=context.config.jdbcUrl) {
+        test {
+                sql """
+                    SELECT * FROM S3 (
+                                        "uri" = "https://${bucket}.${s3_endpoint}/regression/tvf/test_hive_text.text",
+                                        "format" = "hive_text",
+                                        "csv_schema"="k1:int;k2:string;k3:double",
+                                        "resource" = "${resource_name}"
+                                    )  where k1 > 100  order by k3,k2,k1;
+                    """
+                exception "Access denied"
+            }
+    }
+
+    // only have select_priv of view,can select view with resource
+    connect(user=user, password="${pwd}", url=context.config.jdbcUrl) {
+            sql """SELECT * FROM ${viewName};"""
+    }
+
+    try_sql("DROP USER ${user}")
+    sql "drop view if exists ${viewName}"
 }